carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chenliang...@apache.org
Subject [2/3] incubator-carbondata git commit: extract command to common
Date Sat, 24 Dec 2016 14:13:04 GMT
extract command to common

fix

fix testcase

remove redundant QueryTest

fix comment


Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/b50866b3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/b50866b3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/b50866b3

Branch: refs/heads/master
Commit: b50866b3a2e3eb7d704c6fa9f6e6fcdfc5e5de21
Parents: 956eb33
Author: jackylk <jacky.likun@huawei.com>
Authored: Thu Dec 22 15:09:53 2016 +0800
Committer: chenliang613 <chenliang613@apache.org>
Committed: Sat Dec 24 22:10:22 2016 +0800

----------------------------------------------------------------------
 .../core/carbon/AbsoluteTableIdentifier.java    |   6 +
 .../core/carbon/path/CarbonStorePath.java       |   6 +-
 .../core/carbon/path/CarbonTablePath.java       |  35 +-
 .../carbondata/examples/CarbonExample.scala     |   1 +
 .../carbondata/spark/load/CarbonLoaderUtil.java |  19 +-
 .../spark/load/DeleteLoadFolders.java           |  37 +-
 .../carbondata/spark/util/LoadMetadataUtil.java |   7 +-
 .../org/apache/carbondata/api/CarbonStore.scala | 176 ++++++++++
 .../spark/rdd/DataManagementFunc.scala          | 163 ++-------
 .../spark/rdd/CarbonDataRDDFactory.scala        |   4 +-
 .../execution/command/carbonTableSchema.scala   | 349 ++++---------------
 .../apache/spark/sql/hive/CarbonMetastore.scala |   6 +-
 .../spark/load/CarbonLoaderUtilTest.java        |   1 -
 .../spark/rdd/CarbonDataRDDFactory.scala        |   4 +-
 .../execution/command/carbonTableSchema.scala   | 275 +--------------
 .../apache/spark/sql/hive/CarbonMetastore.scala |  10 +-
 .../org/apache/spark/util/CleanFiles.scala      |  15 +-
 .../org/apache/spark/util/Compaction.scala      |  11 +-
 .../apache/spark/util/DeleteSegmentByDate.scala |  14 +-
 .../apache/spark/util/DeleteSegmentById.scala   |  14 +-
 .../org/apache/spark/util/ShowSegments.scala    |  38 +-
 .../org/apache/spark/util/TableAPIUtil.scala    |  23 +-
 .../spark2/src/test/resources/data_alltypes.csv |  10 +
 .../AllDataTypesTestCaseAggregate.scala         |  10 +-
 .../carbondata/CarbonDataSourceSuite.scala      |   3 +-
 .../spark/carbondata/util/QueryTest.scala       |  66 ----
 .../vectorreader/VectorReaderTestCase.scala     |   6 +-
 .../sql/common/util/CarbonSessionTest.scala     |  74 ----
 .../spark/sql/common/util/QueryTest.scala       | 111 +++++-
 .../apache/spark/util/CarbonCommandSuite.scala  |  78 +++++
 .../lcm/status/SegmentStatusManager.java        |   4 +
 31 files changed, 608 insertions(+), 968 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/core/src/main/java/org/apache/carbondata/core/carbon/AbsoluteTableIdentifier.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/AbsoluteTableIdentifier.java b/core/src/main/java/org/apache/carbondata/core/carbon/AbsoluteTableIdentifier.java
index cf7d92f..1424ba5 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/AbsoluteTableIdentifier.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/AbsoluteTableIdentifier.java
@@ -23,6 +23,7 @@ import java.io.Serializable;
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
+import org.apache.carbondata.core.util.CarbonUtil;
 
 /**
  * identifier which will have store path and carbon table identifier
@@ -65,6 +66,11 @@ public class AbsoluteTableIdentifier implements Serializable {
     return carbonTableIdentifier;
   }
 
+  public static AbsoluteTableIdentifier from(String dbName, String tableName) {
+    CarbonTableIdentifier identifier = new CarbonTableIdentifier(dbName, tableName, "");
+    return new AbsoluteTableIdentifier(CarbonUtil.getCarbonStorePath(), identifier);
+  }
+
   public static AbsoluteTableIdentifier fromTablePath(String tablePath) {
     String formattedTablePath = tablePath.replace('\\', '/');
     String[] names = formattedTablePath.split("/");

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonStorePath.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonStorePath.java b/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonStorePath.java
index 214c633..afe4d9a 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonStorePath.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonStorePath.java
@@ -45,10 +45,14 @@ public class CarbonStorePath extends Path {
     CarbonTablePath carbonTablePath = new CarbonTablePath(tableIdentifier,
         storePath + File.separator + tableIdentifier.getDatabaseName() + File.separator
             + tableIdentifier.getTableName());
-
     return carbonTablePath;
   }
 
+  public static CarbonTablePath getCarbonTablePath(String storePath,
+      String dbName, String tableName) {
+    return new CarbonTablePath(storePath, dbName, tableName);
+  }
+
   public static CarbonTablePath getCarbonTablePath(AbsoluteTableIdentifier identifier) {
     CarbonTableIdentifier id = identifier.getCarbonTableIdentifier();
     return new CarbonTablePath(id, identifier.getTablePath());

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonTablePath.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonTablePath.java b/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonTablePath.java
index f90073e..54e7266 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonTablePath.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/path/CarbonTablePath.java
@@ -63,6 +63,12 @@ public class CarbonTablePath extends Path {
     this.tablePath = tablePathString;
   }
 
+  public CarbonTablePath(String storePath, String dbName, String tableName) {
+    super(storePath + File.separator + dbName + File.separator + tableName);
+    this.carbonTableIdentifier = new CarbonTableIdentifier(dbName, tableName, "");
+    this.tablePath = storePath + File.separator + dbName + File.separator + tableName;
+  }
+
   /**
    * The method returns the folder path containing the carbon file.
    *
@@ -249,22 +255,6 @@ public class CarbonTablePath extends Path {
   }
 
   /**
-   * Gets absolute path of data file of given aggregate table
-   *
-   * @param aggTableID          unique aggregate table identifier
-   * @param partitionId         unique partition identifier
-   * @param segmentId           unique partition identifier
-   * @param filePartNo          data file part number
-   * @param factUpdateTimeStamp unique identifier to identify an update
-   * @return absolute path of data file stored in carbon data format
-   */
-  public String getCarbonAggDataFilePath(String aggTableID, String partitionId, String segmentId,
-      Integer filePartNo, Integer taskNo, String factUpdateTimeStamp) {
-    return getAggSegmentDir(aggTableID, partitionId, segmentId) + File.separator
-        + getCarbonDataFileName(filePartNo, taskNo, factUpdateTimeStamp);
-  }
-
-  /**
    * Gets data file name only with out path
    *
    * @param filePartNo          data file part number
@@ -297,15 +287,6 @@ public class CarbonTablePath extends Path {
     return getFactDir() + File.separator + PARTITION_PREFIX + partitionId;
   }
 
-  private String getAggSegmentDir(String aggTableID, String partitionId, String segmentId) {
-    return getAggPartitionDir(aggTableID, partitionId) + File.separator + SEGMENT_PREFIX
-        + segmentId;
-  }
-
-  private String getAggPartitionDir(String aggTableID, String partitionId) {
-    return getAggregateTableDir(aggTableID) + File.separator + PARTITION_PREFIX + partitionId;
-  }
-
   private String getMetaDataDir() {
     return tablePath + File.separator + METADATA_DIR;
   }
@@ -314,10 +295,6 @@ public class CarbonTablePath extends Path {
     return tablePath + File.separator + FACT_DIR;
   }
 
-  private String getAggregateTableDir(String aggTableId) {
-    return tablePath + File.separator + AGGREGATE_TABLE_PREFIX + aggTableId;
-  }
-
   @Override public boolean equals(Object o) {
     if (!(o instanceof CarbonTablePath)) {
       return false;

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonExample.scala
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonExample.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonExample.scala
index c2e135a..273de95 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonExample.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonExample.scala
@@ -21,6 +21,7 @@ import java.io.File
 
 import org.apache.commons.io.FileUtils
 import org.apache.spark.sql.SparkSession
+import org.apache.spark.util.{CleanFiles, ShowSegments}
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.util.CarbonProperties

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
index 492ceee..0d2ab6f 100644
--- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java
@@ -47,7 +47,6 @@ import org.apache.carbondata.core.cache.CacheType;
 import org.apache.carbondata.core.cache.dictionary.Dictionary;
 import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
 import org.apache.carbondata.core.carbon.AbsoluteTableIdentifier;
-import org.apache.carbondata.core.carbon.CarbonDataLoadSchema;
 import org.apache.carbondata.core.carbon.CarbonTableIdentifier;
 import org.apache.carbondata.core.carbon.ColumnIdentifier;
 import org.apache.carbondata.core.carbon.datastore.block.Distributable;
@@ -86,7 +85,6 @@ import com.google.gson.Gson;
 import org.apache.spark.SparkConf;
 import org.apache.spark.util.Utils;
 
-
 public final class CarbonLoaderUtil {
 
   private static final LogService LOGGER =
@@ -449,11 +447,10 @@ public final class CarbonLoaderUtil {
     return status;
   }
 
-  public static void writeLoadMetadata(CarbonDataLoadSchema schema, String databaseName,
-      String tableName, List<LoadMetadataDetails> listOfLoadFolderDetails) throws IOException {
-    CarbonTablePath carbonTablePath = CarbonStorePath
-        .getCarbonTablePath(schema.getCarbonTable().getStorePath(),
-            schema.getCarbonTable().getCarbonTableIdentifier());
+  public static void writeLoadMetadata(String storeLocation, String dbName, String tableName,
+      List<LoadMetadataDetails> listOfLoadFolderDetails) throws IOException {
+    CarbonTablePath carbonTablePath =
+        CarbonStorePath.getCarbonTablePath(storeLocation, dbName, tableName);
     String dataLoadLocation = carbonTablePath.getTableStatusFilePath();
 
     DataOutputStream dataOutputStream;
@@ -464,11 +461,9 @@ public final class CarbonLoaderUtil {
         new AtomicFileOperationsImpl(dataLoadLocation, FileFactory.getFileType(dataLoadLocation));
 
     try {
-
       dataOutputStream = writeOperation.openForWrite(FileWriteOperation.OVERWRITE);
       brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
               Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
-
       String metadataInstance = gsonObjectToWrite.toJson(listOfLoadFolderDetails.toArray());
       brWriter.write(metadataInstance);
     } finally {
@@ -478,12 +473,10 @@ public final class CarbonLoaderUtil {
         }
       } catch (Exception e) {
         LOGGER.error("error in  flushing ");
-
       }
       CarbonUtil.closeStreams(brWriter);
       writeOperation.close();
     }
-
   }
 
   public static String readCurrentTime() {
@@ -495,10 +488,10 @@ public final class CarbonLoaderUtil {
     return date;
   }
 
-  public static String extractLoadMetadataFileLocation(CarbonLoadModel loadModel) {
+  public static String extractLoadMetadataFileLocation(String dbName, String tableName) {
     CarbonTable carbonTable =
         org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance()
-            .getCarbonTable(loadModel.getDatabaseName() + '_' + loadModel.getTableName());
+            .getCarbonTable(dbName + '_' + tableName);
     return carbonTable.getMetaDataFilepath();
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java
index 2b3979f..0663abc 100644
--- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java
@@ -37,6 +37,7 @@ import java.util.List;
 
 import org.apache.carbondata.common.logging.LogService;
 import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.carbon.CarbonTableIdentifier;
 import org.apache.carbondata.core.carbon.path.CarbonStorePath;
 import org.apache.carbondata.core.carbon.path.CarbonTablePath;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
@@ -45,7 +46,6 @@ import org.apache.carbondata.core.datastorage.store.filesystem.CarbonFileFilter;
 import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
 import org.apache.carbondata.core.load.LoadMetadataDetails;
 import org.apache.carbondata.core.util.CarbonProperties;
-import org.apache.carbondata.processing.model.CarbonLoadModel;
 
 public final class DeleteLoadFolders {
 
@@ -58,23 +58,13 @@ public final class DeleteLoadFolders {
 
   /**
    * returns segment path
-   *
-   * @param loadModel
-   * @param storeLocation
-   * @param partitionId
-   * @param oneLoad
-   * @return
    */
-  private static String getSegmentPath(CarbonLoadModel loadModel, String storeLocation,
+  private static String getSegmentPath(String dbName, String tableName, String storeLocation,
       int partitionId, LoadMetadataDetails oneLoad) {
-
-    String path = null;
+    CarbonTablePath carbon = new CarbonStorePath(storeLocation).getCarbonTablePath(
+        new CarbonTableIdentifier(dbName, tableName, ""));
     String segmentId = oneLoad.getLoadName();
-
-    path = new CarbonStorePath(storeLocation).getCarbonTablePath(
-        loadModel.getCarbonDataLoadSchema().getCarbonTable().getCarbonTableIdentifier())
-        .getCarbonDataDirectoryPath("" + partitionId, segmentId);
-    return path;
+    return carbon.getCarbonDataDirectoryPath("" + partitionId, segmentId);
   }
 
   private static boolean physicalFactAndMeasureMetadataDeletion(String path) {
@@ -221,32 +211,21 @@ public final class DeleteLoadFolders {
     }
   }
 
-  /**
-   * @param loadModel
-   * @param storeLocation
-   * @param isForceDelete
-   * @param details
-   * @return
-   *
-   */
-  public static boolean deleteLoadFoldersFromFileSystem(CarbonLoadModel loadModel,
+  public static boolean deleteLoadFoldersFromFileSystem(String dbName, String tableName,
       String storeLocation, boolean isForceDelete, LoadMetadataDetails[] details) {
     List<LoadMetadataDetails> deletedLoads =
         new ArrayList<LoadMetadataDetails>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
-
     boolean isDeleted = false;
-
     if (details != null && details.length != 0) {
       for (LoadMetadataDetails oneLoad : details) {
         if (checkIfLoadCanBeDeleted(oneLoad, isForceDelete)) {
-          String path = getSegmentPath(loadModel, storeLocation, 0, oneLoad);
+          String path = getSegmentPath(dbName, tableName, storeLocation, 0, oneLoad);
           boolean deletionStatus = physicalFactAndMeasureMetadataDeletion(path);
           if (deletionStatus) {
             isDeleted = true;
             oneLoad.setVisibility("false");
             deletedLoads.add(oneLoad);
-            LOGGER.info("Info: " +
-                " Deleted the load " + oneLoad.getLoadName());
+            LOGGER.info("Info: Deleted the load " + oneLoad.getLoadName());
           }
         }
       }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/LoadMetadataUtil.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/LoadMetadataUtil.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/LoadMetadataUtil.java
index 11cf9f8..1c21fab 100644
--- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/LoadMetadataUtil.java
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/LoadMetadataUtil.java
@@ -28,20 +28,19 @@
  */
 package org.apache.carbondata.spark.util;
 
+import org.apache.carbondata.core.carbon.metadata.CarbonMetadata;
 import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.load.LoadMetadataDetails;
 import org.apache.carbondata.lcm.status.SegmentStatusManager;
-import org.apache.carbondata.processing.model.CarbonLoadModel;
 
 public final class LoadMetadataUtil {
   private LoadMetadataUtil() {
 
   }
 
-  public static boolean isLoadDeletionRequired(CarbonLoadModel loadModel) {
-    CarbonTable table = org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance()
-        .getCarbonTable(loadModel.getDatabaseName() + '_' + loadModel.getTableName());
+  public static boolean isLoadDeletionRequired(String dbName, String tableName) {
+    CarbonTable table = CarbonMetadata.getInstance().getCarbonTable(dbName + '_' + tableName);
 
     String metaDataLocation = table.getMetaDataFilepath();
     LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(metaDataLocation);

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
new file mode 100644
index 0000000..6fb2df1
--- /dev/null
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.api
+
+import java.lang.Long
+import java.text.SimpleDateFormat
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+import org.apache.spark.sql.types.TimestampType
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.carbon.AbsoluteTableIdentifier
+import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.lcm.status.SegmentStatusManager
+import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
+import org.apache.carbondata.spark.rdd.DataManagementFunc
+
+object CarbonStore {
+  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+
+  def showSegments(
+      dbName: String,
+      tableName: String,
+      limit: Option[String]): Seq[Row] = {
+    val tableUniqueName = dbName + "_" + tableName
+    val carbonTable = CarbonMetadata.getInstance().getCarbonTable(tableUniqueName)
+    val path = carbonTable.getMetaDataFilepath
+    val loadMetadataDetailsArray = SegmentStatusManager.readLoadMetadata(path)
+    if (loadMetadataDetailsArray.nonEmpty) {
+      val parser = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP)
+      var loadMetadataDetailsSortedArray = loadMetadataDetailsArray.sortWith { (l1, l2) =>
+        java.lang.Double.parseDouble(l1.getLoadName) > java.lang.Double.parseDouble(l2.getLoadName)
+      }
+      if (limit.isDefined) {
+        loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray
+          .filter(load => load.getVisibility.equalsIgnoreCase("true"))
+        val limitLoads = limit.get
+        try {
+          val lim = Integer.parseInt(limitLoads)
+          loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray.slice(0, lim)
+        } catch {
+          case _: NumberFormatException => sys.error(s" Entered limit is not a valid Number")
+        }
+      }
+
+      loadMetadataDetailsSortedArray
+          .filter(_.getVisibility.equalsIgnoreCase("true"))
+          .map { load =>
+            Row(
+              load.getLoadName,
+              load.getLoadStatus,
+              new java.sql.Timestamp(parser.parse(load.getLoadStartTime).getTime),
+              new java.sql.Timestamp(parser.parse(load.getTimestamp).getTime)
+            )
+          }.toSeq
+    } else {
+      Seq.empty
+    }
+  }
+
+  def cleanFiles(
+      dbName: String,
+      tableName: String,
+      storePath: String): Unit = {
+    LOGGER.audit(s"The clean files request has been received for $dbName.$tableName")
+    try {
+      DataManagementFunc.cleanFiles(dbName, tableName, storePath)
+      LOGGER.audit(s"Clean files operation is success for $dbName.$tableName.")
+    } catch {
+      case ex: Exception =>
+        sys.error(ex.getMessage)
+    }
+    Seq.empty
+  }
+
+  // validates load ids
+  private def validateLoadIds(loadids: Seq[String]): Unit = {
+    if (loadids.isEmpty) {
+      val errorMessage = "Error: Segment id(s) should not be empty."
+      throw new MalformedCarbonCommandException(errorMessage)
+    }
+  }
+
+  def deleteLoadById(
+      loadids: Seq[String],
+      dbName: String,
+      tableName: String): Unit = {
+
+    LOGGER.audit(s"Delete segment by Id request has been received for $dbName.$tableName")
+    validateLoadIds(loadids)
+
+    val carbonTable = CarbonMetadata.getInstance().getCarbonTable(dbName + '_' + tableName)
+    val path = carbonTable.getMetaDataFilepath
+
+    try {
+      val invalidLoadIds = SegmentStatusManager.updateDeletionStatus(
+        carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala
+      if (invalidLoadIds.isEmpty) {
+        LOGGER.audit(s"Delete segment by Id is successfull for $dbName.$tableName.")
+      } else {
+        sys.error(s"Delete segment by Id is failed. Invalid ID is: ${invalidLoadIds.mkString(",")}")
+      }
+    } catch {
+      case ex: Exception =>
+        sys.error(ex.getMessage)
+    }
+    Seq.empty
+  }
+
+  def deleteLoadByDate(
+      timestamp: String,
+      dbName: String,
+      tableName: String): Unit = {
+    LOGGER.audit(s"Delete segment by Id request has been received for $dbName.$tableName")
+
+    val time = validateTimeFormat(timestamp)
+    val carbonTable = CarbonMetadata.getInstance().getCarbonTable(dbName + '_' + tableName)
+    val path = carbonTable.getMetaDataFilepath
+
+    try {
+      val invalidLoadTimestamps =
+        SegmentStatusManager.updateDeletionStatus(
+          carbonTable.getAbsoluteTableIdentifier,
+          timestamp,
+          path,
+          time).asScala
+      if (invalidLoadTimestamps.isEmpty) {
+        LOGGER.audit(s"Delete segment by date is successful for $dbName.$tableName.")
+      } else {
+        sys.error("Delete segment by date is failed. No matching segment found.")
+      }
+    } catch {
+      case ex: Exception =>
+        sys.error(ex.getMessage)
+    }
+  }
+
+  // this function is for test only
+  def isSegmentValid(
+      dbName: String,
+      tableName: String,
+      segmentId: String): Boolean = {
+    val identifier = AbsoluteTableIdentifier.from(dbName, tableName)
+    val status = SegmentStatusManager.getSegmentStatus(identifier)
+    status.isValid(segmentId)
+  }
+
+  private def validateTimeFormat(timestamp: String): Long = {
+    val timeObj = Cast(Literal(timestamp), TimestampType).eval()
+    if (null == timeObj) {
+      val errorMessage = "Error: Invalid load start time format: " + timestamp
+      throw new MalformedCarbonCommandException(errorMessage)
+    }
+    timeObj.asInstanceOf[Long]
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
index 28a9140..c2f06a4 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
@@ -21,22 +21,18 @@ import java.util
 import java.util.concurrent._
 
 import scala.collection.JavaConverters._
-import scala.collection.mutable.ListBuffer
 
-import org.apache.spark.SparkContext
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.execution.command.{CompactionCallableModel, CompactionModel}
 
 import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.carbon.CarbonDataLoadSchema
-import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
+import org.apache.carbondata.core.carbon.{CarbonDataLoadSchema, CarbonTableIdentifier}
 import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.load.LoadMetadataDetails
 import org.apache.carbondata.lcm.locks.{CarbonLockFactory, CarbonLockUtil, LockUsage}
 import org.apache.carbondata.lcm.status.SegmentStatusManager
 import org.apache.carbondata.processing.model.CarbonLoadModel
-import org.apache.carbondata.spark._
 import org.apache.carbondata.spark.load._
 import org.apache.carbondata.spark.merger.{CarbonDataMergerUtil, CompactionCallable, CompactionType}
 import org.apache.carbondata.spark.util.{CommonUtil, LoadMetadataUtil}
@@ -48,104 +44,6 @@ object DataManagementFunc {
 
   private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
 
-  def deleteLoadByDate(
-      sqlContext: SQLContext,
-      schema: CarbonDataLoadSchema,
-      databaseName: String,
-      tableName: String,
-      storePath: String,
-      dateField: String,
-      dateFieldActualName: String,
-      dateValue: String) {
-
-    val sc = sqlContext
-    // Delete the records based on data
-    val table = CarbonMetadata.getInstance.getCarbonTable(databaseName + "_" + tableName)
-    val loadMetadataDetailsArray =
-      SegmentStatusManager.readLoadMetadata(table.getMetaDataFilepath).toList
-    val resultMap = new CarbonDeleteLoadByDateRDD(
-      sc.sparkContext,
-      new DeletedLoadResultImpl(),
-      databaseName,
-      table.getDatabaseName,
-      dateField,
-      dateFieldActualName,
-      dateValue,
-      table.getFactTableName,
-      tableName,
-      storePath,
-      loadMetadataDetailsArray).collect.groupBy(_._1)
-
-    var updatedLoadMetadataDetailsList = new ListBuffer[LoadMetadataDetails]()
-    if (resultMap.nonEmpty) {
-      if (resultMap.size == 1) {
-        if (resultMap.contains("")) {
-          LOGGER.error("Delete by Date request is failed")
-          sys.error("Delete by Date request is failed, potential causes " +
-              "Empty store or Invalid column type, For more details please refer logs.")
-        }
-      }
-      val updatedloadMetadataDetails = loadMetadataDetailsArray.map { elem => {
-        var statusList = resultMap.get(elem.getLoadName)
-        // check for the merged load folder.
-        if (statusList.isEmpty && null != elem.getMergedLoadName) {
-          statusList = resultMap.get(elem.getMergedLoadName)
-        }
-
-        if (statusList.isDefined) {
-          elem.setModificationOrdeletionTimesStamp(CarbonLoaderUtil.readCurrentTime())
-          // if atleast on CarbonCommonConstants.MARKED_FOR_UPDATE status exist,
-          // use MARKED_FOR_UPDATE
-          if (statusList.get
-              .forall(status => status._2 == CarbonCommonConstants.MARKED_FOR_DELETE)) {
-            elem.setLoadStatus(CarbonCommonConstants.MARKED_FOR_DELETE)
-          } else {
-            elem.setLoadStatus(CarbonCommonConstants.MARKED_FOR_UPDATE)
-            updatedLoadMetadataDetailsList += elem
-          }
-          elem
-        } else {
-          elem
-        }
-      }
-
-      }
-
-      // Save the load metadata
-      val carbonLock = CarbonLockFactory
-          .getCarbonLockObj(table.getAbsoluteTableIdentifier.getCarbonTableIdentifier,
-            LockUsage.METADATA_LOCK
-          )
-      try {
-        if (carbonLock.lockWithRetries()) {
-          LOGGER.info("Successfully got the table metadata file lock")
-          if (updatedLoadMetadataDetailsList.nonEmpty) {
-            // TODO: Load Aggregate tables after retention.
-          }
-
-          // write
-          CarbonLoaderUtil.writeLoadMetadata(
-            schema,
-            databaseName,
-            table.getDatabaseName,
-            updatedloadMetadataDetails.asJava
-          )
-        }
-      } finally {
-        if (carbonLock.unlock()) {
-          LOGGER.info("unlock the table metadata file successfully")
-        } else {
-          LOGGER.error("Unable to unlock the metadata lock")
-        }
-      }
-    } else {
-      LOGGER.error("Delete by Date request is failed")
-      LOGGER.audit(s"The delete load by date is failed for $databaseName.$tableName")
-      sys.error("Delete by Date request is failed, potential causes " +
-          "Empty store or Invalid column type, For more details please refer logs.")
-    }
-  }
-
   def executeCompaction(carbonLoadModel: CarbonLoadModel,
       storePath: String,
       compactionModel: CompactionModel,
@@ -224,7 +122,7 @@ object DataManagementFunc {
    *
    * @param futureList
    */
-  def scanSegmentsAndSubmitJob(futureList: util.List[Future[Void]],
+  private def scanSegmentsAndSubmitJob(futureList: util.List[Future[Void]],
       loadsToMerge: util
       .List[LoadMetadataDetails],
       executor: ExecutorService,
@@ -289,21 +187,29 @@ object DataManagementFunc {
   }
 
   def deleteLoadsAndUpdateMetadata(
-      carbonLoadModel: CarbonLoadModel,
-      table: CarbonTable,
+      dbName: String,
+      tableName: String,
       storePath: String,
       isForceDeletion: Boolean): Unit = {
-    if (LoadMetadataUtil.isLoadDeletionRequired(carbonLoadModel)) {
-      val loadMetadataFilePath = CarbonLoaderUtil
-          .extractLoadMetadataFileLocation(carbonLoadModel)
+    if (LoadMetadataUtil.isLoadDeletionRequired(dbName, tableName)) {
+      val loadMetadataFilePath =
+        CarbonLoaderUtil.extractLoadMetadataFileLocation(dbName, tableName)
       val details = SegmentStatusManager.readLoadMetadata(loadMetadataFilePath)
-      val carbonTableStatusLock = CarbonLockFactory
-          .getCarbonLockObj(table.getAbsoluteTableIdentifier.getCarbonTableIdentifier,
-            LockUsage.TABLE_STATUS_LOCK)
+      val carbonTableStatusLock =
+        CarbonLockFactory.getCarbonLockObj(
+          new CarbonTableIdentifier(dbName, tableName, ""),
+          LockUsage.TABLE_STATUS_LOCK
+        )
 
       // Delete marked loads
-      val isUpdationRequired = DeleteLoadFolders
-          .deleteLoadFoldersFromFileSystem(carbonLoadModel, storePath, isForceDeletion, details)
+      val isUpdationRequired =
+        DeleteLoadFolders.deleteLoadFoldersFromFileSystem(
+          dbName,
+          tableName,
+          storePath,
+          isForceDeletion,
+          details
+        )
 
       if (isUpdationRequired) {
         try {
@@ -318,14 +224,10 @@ object DataManagementFunc {
             val latestStatus = CarbonLoaderUtil
                 .updateLoadMetadataFromOldToNew(details, latestMetadata)
 
-            CarbonLoaderUtil.writeLoadMetadata(
-              carbonLoadModel.getCarbonDataLoadSchema,
-              carbonLoadModel.getDatabaseName,
-              carbonLoadModel.getTableName, latestStatus)
+            CarbonLoaderUtil.writeLoadMetadata(storePath, dbName, tableName, latestStatus)
           } else {
             val errorMsg = "Clean files request is failed for " +
-                s"${ carbonLoadModel.getDatabaseName }." +
-                s"${ carbonLoadModel.getTableName }" +
+                s"$dbName.$tableName" +
                 ". Not able to acquire the table status lock due to other operation " +
                 "running in the background."
             LOGGER.audit(errorMsg)
@@ -340,29 +242,24 @@ object DataManagementFunc {
   }
 
   def cleanFiles(
-      sc: SparkContext,
-      carbonLoadModel: CarbonLoadModel,
-      storePath: String) {
-    val table = CarbonMetadata.getInstance.getCarbonTable(
-      carbonLoadModel.getDatabaseName + "_" + carbonLoadModel.getTableName)
-    val carbonCleanFilesLock = CarbonLockFactory.getCarbonLockObj(
-      table.getAbsoluteTableIdentifier.getCarbonTableIdentifier, LockUsage.CLEAN_FILES_LOCK)
+      dbName: String,
+      tableName: String,
+      storePath: String): Unit = {
+    val identifier = new CarbonTableIdentifier(dbName, tableName, "")
+    val carbonCleanFilesLock =
+      CarbonLockFactory.getCarbonLockObj(identifier, LockUsage.CLEAN_FILES_LOCK)
     try {
       if (carbonCleanFilesLock.lockWithRetries()) {
         LOGGER.info("Clean files lock has been successfully acquired.")
-        deleteLoadsAndUpdateMetadata(carbonLoadModel,
-          table,
-          storePath,
-          isForceDeletion = true)
+        deleteLoadsAndUpdateMetadata(dbName, tableName, storePath, isForceDeletion = true)
       } else {
         val errorMsg = "Clean files request is failed for " +
-            s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }" +
+            s"$dbName.$tableName" +
             ". Not able to acquire the clean files lock due to another clean files " +
             "operation is running in the background."
         LOGGER.audit(errorMsg)
         LOGGER.error(errorMsg)
         throw new Exception(errorMsg + " Please try after some time.")
-
       }
     } finally {
       CarbonLockUtil.fileUnlock(carbonCleanFilesLock, LockUsage.CLEAN_FILES_LOCK)

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 53a5f67..93194c8 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -433,8 +433,8 @@ object CarbonDataRDDFactory {
                      s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
       }
       // Check if any load need to be deleted before loading new data
-      DataManagementFunc.deleteLoadsAndUpdateMetadata(carbonLoadModel, carbonTable, storePath,
-        isForceDeletion = false)
+      DataManagementFunc.deleteLoadsAndUpdateMetadata(carbonLoadModel.getDatabaseName,
+        carbonLoadModel.getTableName, storePath, isForceDeletion = false)
       if (null == carbonLoadModel.getLoadMetadataDetails) {
         CommonUtil.readLoadMetadataDetails(carbonLoadModel, storePath)
       }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index 041b65f..d82290e 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -18,14 +18,13 @@
 package org.apache.spark.sql.execution.command
 
 import java.io.File
-import java.text.SimpleDateFormat
 
 import scala.collection.JavaConverters._
 import scala.language.implicitConversions
 
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Cast, Literal}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Literal}
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.{RunnableCommand, SparkPlan}
 import org.apache.spark.sql.hive.CarbonMetastore
@@ -33,9 +32,9 @@ import org.apache.spark.sql.types.TimestampType
 import org.apache.spark.util.FileUtils
 import org.codehaus.jackson.map.ObjectMapper
 
+import org.apache.carbondata.api.CarbonStore
 import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.carbon.{CarbonDataLoadSchema, CarbonTableIdentifier}
-import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
 import org.apache.carbondata.core.carbon.metadata.encoder.Encoding
 import org.apache.carbondata.core.carbon.metadata.schema.table.{CarbonTable, TableInfo}
 import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension
@@ -49,9 +48,23 @@ import org.apache.carbondata.processing.constants.TableOptionConstant
 import org.apache.carbondata.processing.etl.DataLoadingException
 import org.apache.carbondata.processing.model.CarbonLoadModel
 import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
-import org.apache.carbondata.spark.rdd.{CarbonDataRDDFactory, DataManagementFunc, DataManagementFunc$, DictionaryLoadModel}
+import org.apache.carbondata.spark.rdd.{CarbonDataRDDFactory, DictionaryLoadModel}
 import org.apache.carbondata.spark.util.{CarbonScalaUtil, GlobalDictionaryUtil}
 
+object Checker {
+  def validateTableExists(
+      dbName: Option[String],
+      tableName: String,
+      sqlContext: SQLContext): Unit = {
+    val identifier = TableIdentifier(tableName, dbName)
+    if (!CarbonEnv.get.carbonMetastore.tableExists(identifier)(sqlContext)) {
+      val err = s"table $dbName.$tableName not found"
+      LogServiceFactory.getLogService(this.getClass.getName).error(err)
+      throw new IllegalArgumentException(err)
+    }
+  }
+}
+
 /**
  * Command for the compaction in alter table command
  *
@@ -179,61 +192,14 @@ private[sql] case class DeleteLoadsById(
     databaseNameOp: Option[String],
     tableName: String) extends RunnableCommand {
 
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
   def run(sqlContext: SQLContext): Seq[Row] = {
-
-    val databaseName = getDB.getDatabaseName(databaseNameOp, sqlContext)
-    LOGGER.audit(s"Delete segment by Id request has been received for $databaseName.$tableName")
-
-    // validate load ids first
-    validateLoadIds
-    val dbName = getDB.getDatabaseName(databaseNameOp, sqlContext)
-
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore.lookupRelation1(
-      identifier, None)(sqlContext).asInstanceOf[CarbonRelation]
-    if (relation == null) {
-      LOGGER.audit(s"Delete segment by Id is failed. Table $dbName.$tableName does not exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-
-    val carbonTable = CarbonMetadata.getInstance().getCarbonTable(dbName + '_' + tableName)
-
-    if (null == carbonTable) {
-      CarbonEnv.get.carbonMetastore
-        .lookupRelation1(identifier, None)(sqlContext).asInstanceOf[CarbonRelation]
-    }
-    val path = carbonTable.getMetaDataFilepath
-
-    try {
-      val invalidLoadIds = SegmentStatusManager.updateDeletionStatus(
-        carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala
-
-      if (invalidLoadIds.isEmpty) {
-
-        LOGGER.audit(s"Delete segment by Id is successfull for $databaseName.$tableName.")
-      }
-      else {
-        sys.error("Delete segment by Id is failed. Invalid ID is:" +
-                  s" ${ invalidLoadIds.mkString(",") }")
-      }
-    } catch {
-      case ex: Exception =>
-        sys.error(ex.getMessage)
-    }
-
+    Checker.validateTableExists(databaseNameOp, tableName, sqlContext)
+    CarbonStore.deleteLoadById(
+      loadids,
+      getDB.getDatabaseName(databaseNameOp, sqlContext),
+      tableName
+    )
     Seq.empty
-
-  }
-
-  // validates load ids
-  private def validateLoadIds: Unit = {
-    if (loadids.isEmpty) {
-      val errorMessage = "Error: Segment id(s) should not be empty."
-      throw new MalformedCarbonCommandException(errorMessage)
-
-    }
   }
 }
 
@@ -243,90 +209,14 @@ private[sql] case class DeleteLoadsByLoadDate(
     dateField: String,
     loadDate: String) extends RunnableCommand {
 
-  val LOGGER = LogServiceFactory.getLogService("org.apache.spark.sql.tablemodel.tableSchema")
-
   def run(sqlContext: SQLContext): Seq[Row] = {
-
-    LOGGER.audit("The delete segment by load date request has been received.")
-    val dbName = getDB.getDatabaseName(databaseNameOp, sqlContext)
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore
-      .lookupRelation1(identifier, None)(sqlContext).asInstanceOf[CarbonRelation]
-    if (relation == null) {
-      LOGGER
-        .audit(s"Delete segment by load date is failed. Table $dbName.$tableName does not " +
-               s"exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-
-    val timeObj = Cast(Literal(loadDate), TimestampType).eval()
-    if (null == timeObj) {
-      val errorMessage = "Error: Invalid load start time format " + loadDate
-      throw new MalformedCarbonCommandException(errorMessage)
-    }
-
-    val carbonTable = org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance()
-      .getCarbonTable(dbName + '_' + tableName)
-    if (null == carbonTable) {
-      var relation = CarbonEnv.get.carbonMetastore
-        .lookupRelation1(identifier, None)(sqlContext).asInstanceOf[CarbonRelation]
-    }
-    val path = carbonTable.getMetaDataFilepath()
-
-    try {
-      val invalidLoadTimestamps = SegmentStatusManager.updateDeletionStatus(
-        carbonTable.getAbsoluteTableIdentifier, loadDate, path,
-        timeObj.asInstanceOf[java.lang.Long]).asScala
-      if (invalidLoadTimestamps.isEmpty) {
-        LOGGER.audit(s"Delete segment by date is successfull for $dbName.$tableName.")
-      }
-      else {
-        sys.error("Delete segment by date is failed. No matching segment found.")
-      }
-    } catch {
-      case ex: Exception =>
-        sys.error(ex.getMessage)
-    }
+    Checker.validateTableExists(databaseNameOp, tableName, sqlContext)
+    CarbonStore.deleteLoadByDate(
+      loadDate,
+      getDB.getDatabaseName(databaseNameOp, sqlContext),
+      tableName
+    )
     Seq.empty
-
-  }
-
-}
-
-object LoadTable {
-
-  def updateTableMetadata(carbonLoadModel: CarbonLoadModel,
-      sqlContext: SQLContext,
-      model: DictionaryLoadModel,
-      noDictDimension: Array[CarbonDimension]): Unit = {
-
-    val carbonTablePath = CarbonStorePath.getCarbonTablePath(model.hdfsLocation,
-      model.table)
-    val schemaFilePath = carbonTablePath.getSchemaFilePath
-
-    // read TableInfo
-    val tableInfo = CarbonMetastore.readSchemaFileToThriftTable(schemaFilePath)
-
-    // modify TableInfo
-    val columns = tableInfo.getFact_table.getTable_columns
-    for (i <- 0 until columns.size) {
-      if (noDictDimension.exists(x => columns.get(i).getColumn_id.equals(x.getColumnId))) {
-        columns.get(i).encoders.remove(org.apache.carbondata.format.Encoding.DICTIONARY)
-      }
-    }
-
-    // write TableInfo
-    CarbonMetastore.writeThriftTableToSchemaFile(schemaFilePath, tableInfo)
-
-    // update Metadata
-    val catalog = CarbonEnv.get.carbonMetastore
-    catalog.updateMetadataByThriftTable(schemaFilePath, tableInfo,
-      model.table.getDatabaseName, model.table.getTableName, carbonLoadModel.getStorePath)
-
-    // update CarbonDataLoadSchema
-    val carbonTable = catalog.lookupRelation1(Option(model.table.getDatabaseName),
-      model.table.getTableName)(sqlContext).asInstanceOf[CarbonRelation].tableMeta.carbonTable
-    carbonLoadModel.setCarbonDataLoadSchema(new CarbonDataLoadSchema(carbonTable))
   }
 
 }
@@ -511,7 +401,7 @@ case class LoadTable(
         carbonLoadModel.setCsvHeader(fileHeader)
         carbonLoadModel.setColDictFilePath(columnDict)
         carbonLoadModel.setDirectLoad(true)
-        GlobalDictionaryUtil.updateTableMetadataFunc = LoadTable.updateTableMetadata
+        GlobalDictionaryUtil.updateTableMetadataFunc = updateTableMetadata
         GlobalDictionaryUtil
           .generateGlobalDictionary(sqlContext, carbonLoadModel, relation.tableMeta.storePath,
             dataFrame)
@@ -565,6 +455,40 @@ case class LoadTable(
     Seq.empty
   }
 
+  private def updateTableMetadata(carbonLoadModel: CarbonLoadModel,
+      sqlContext: SQLContext,
+      model: DictionaryLoadModel,
+      noDictDimension: Array[CarbonDimension]): Unit = {
+
+    val carbonTablePath = CarbonStorePath.getCarbonTablePath(model.hdfsLocation,
+      model.table)
+    val schemaFilePath = carbonTablePath.getSchemaFilePath
+
+    // read TableInfo
+    val tableInfo = CarbonMetastore.readSchemaFileToThriftTable(schemaFilePath)
+
+    // modify TableInfo
+    val columns = tableInfo.getFact_table.getTable_columns
+    for (i <- 0 until columns.size) {
+      if (noDictDimension.exists(x => columns.get(i).getColumn_id.equals(x.getColumnId))) {
+        columns.get(i).encoders.remove(org.apache.carbondata.format.Encoding.DICTIONARY)
+      }
+    }
+
+    // write TableInfo
+    CarbonMetastore.writeThriftTableToSchemaFile(schemaFilePath, tableInfo)
+
+    // update Metadata
+    val catalog = CarbonEnv.get.carbonMetastore
+    catalog.updateMetadataByThriftTable(schemaFilePath, tableInfo,
+      model.table.getDatabaseName, model.table.getTableName, carbonLoadModel.getStorePath)
+
+    // update CarbonDataLoadSchema
+    val carbonTable = catalog.lookupRelation1(Option(model.table.getDatabaseName),
+      model.table.getTableName)(sqlContext).asInstanceOf[CarbonRelation].tableMeta.carbonTable
+    carbonLoadModel.setCarbonDataLoadSchema(new CarbonDataLoadSchema(carbonTable))
+  }
+
   private def validateDateFormat(dateFormat: String, table: CarbonTable): Unit = {
     val dimensions = table.getDimensionByTableName(tableName).asScala
     if (dateFormat != null) {
@@ -651,60 +575,14 @@ private[sql] case class ShowLoads(
     limit: Option[String],
     override val output: Seq[Attribute]) extends RunnableCommand {
 
-
   override def run(sqlContext: SQLContext): Seq[Row] = {
-    val databaseName = getDB.getDatabaseName(databaseNameOp, sqlContext)
-    val tableUniqueName = databaseName + "_" + tableName
-    // Here using checkSchemasModifiedTimeAndReloadTables in tableExists to reload metadata if
-    // schema is changed by other process, so that tableInfoMap woulb be refilled.
-    val tableExists = CarbonEnv.get.carbonMetastore
-      .tableExists(TableIdentifier(tableName, databaseNameOp))(sqlContext)
-    if (!tableExists) {
-      sys.error(s"$databaseName.$tableName is not found")
-    }
-    val carbonTable = org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance()
-      .getCarbonTable(tableUniqueName)
-    if (carbonTable == null) {
-      sys.error(s"$databaseName.$tableName is not found")
-    }
-    val path = carbonTable.getMetaDataFilepath
-    val loadMetadataDetailsArray = SegmentStatusManager.readLoadMetadata(path)
-    if (loadMetadataDetailsArray.nonEmpty) {
-
-      val parser = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP)
-
-      var loadMetadataDetailsSortedArray = loadMetadataDetailsArray.sortWith(
-        (l1, l2) => java.lang.Double.parseDouble(l1.getLoadName) > java.lang.Double
-          .parseDouble(l2.getLoadName)
-      )
-
-
-      if (limit.isDefined) {
-        loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray
-          .filter(load => load.getVisibility.equalsIgnoreCase("true"))
-        val limitLoads = limit.get
-        try {
-          val lim = Integer.parseInt(limitLoads)
-          loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray.slice(0, lim)
-        } catch {
-          case ex: NumberFormatException => sys.error(s" Entered limit is not a valid Number")
-        }
-
-      }
-
-      loadMetadataDetailsSortedArray.filter(load => load.getVisibility.equalsIgnoreCase("true"))
-        .map(load =>
-          Row(
-            load.getLoadName,
-            load.getLoadStatus,
-            new java.sql.Timestamp(parser.parse(load.getLoadStartTime).getTime),
-            new java.sql.Timestamp(parser.parse(load.getTimestamp).getTime))).toSeq
-    } else {
-      Seq.empty
-
-    }
+    Checker.validateTableExists(databaseNameOp, tableName, sqlContext)
+    CarbonStore.showSegments(
+      getDB.getDatabaseName(databaseNameOp, sqlContext),
+      tableName,
+      limit
+    )
   }
-
 }
 
 private[sql] case class DescribeCommandFormatted(
@@ -786,92 +664,17 @@ private[sql] case class DescribeCommandFormatted(
   }
 }
 
-private[sql] case class DeleteLoadByDate(
-    databaseNameOp: Option[String],
-    tableName: String,
-    dateField: String,
-    dateValue: String
-) extends RunnableCommand {
-
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  def run(sqlContext: SQLContext): Seq[Row] = {
-    val dbName = getDB.getDatabaseName(databaseNameOp, sqlContext)
-    LOGGER.audit(s"The delete load by date request has been received for $dbName.$tableName")
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore
-      .lookupRelation1(identifier)(sqlContext).asInstanceOf[CarbonRelation]
-    var level: String = ""
-    val carbonTable = org.apache.carbondata.core.carbon.metadata.CarbonMetadata
-      .getInstance().getCarbonTable(dbName + '_' + tableName)
-    if (relation == null) {
-      LOGGER.audit(s"The delete load by date is failed. Table $dbName.$tableName does not exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-    val matches: Seq[AttributeReference] = relation.dimensionsAttr.filter(
-      filter => filter.name.equalsIgnoreCase(dateField) &&
-                filter.dataType.isInstanceOf[TimestampType]).toList
-    if (matches.isEmpty) {
-      LOGGER.audit("The delete load by date is failed. " +
-                   s"Table $dbName.$tableName does not contain date field: $dateField")
-      sys.error(s"Table $dbName.$tableName does not contain date field $dateField")
-    } else {
-      level = matches.asJava.get(0).name
-    }
-    val actualColName = relation.metaData.carbonTable.getDimensionByName(tableName, level)
-      .getColName
-    DataManagementFunc.deleteLoadByDate(
-      sqlContext,
-      new CarbonDataLoadSchema(carbonTable),
-      dbName,
-      tableName,
-      CarbonEnv.get.carbonMetastore.storePath,
-      level,
-      actualColName,
-      dateValue)
-    LOGGER.audit(s"The delete load by date $dateValue is successful for $dbName.$tableName.")
-    Seq.empty
-  }
-
-}
-
 private[sql] case class CleanFiles(
     databaseNameOp: Option[String],
     tableName: String) extends RunnableCommand {
 
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
   def run(sqlContext: SQLContext): Seq[Row] = {
-    val dbName = getDB.getDatabaseName(databaseNameOp, sqlContext)
-    LOGGER.audit(s"The clean files request has been received for $dbName.$tableName")
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore
-      .lookupRelation1(identifier)(sqlContext).
-      asInstanceOf[CarbonRelation]
-    if (relation == null) {
-      LOGGER.audit(s"The clean files request is failed. Table $dbName.$tableName does not exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-
-    val carbonLoadModel = new CarbonLoadModel()
-    carbonLoadModel.setTableName(relation.tableMeta.carbonTableIdentifier.getTableName)
-    carbonLoadModel.setDatabaseName(relation.tableMeta.carbonTableIdentifier.getDatabaseName)
-    val table = relation.tableMeta.carbonTable
-    carbonLoadModel.setAggTables(table.getAggregateTablesName.asScala.toArray)
-    carbonLoadModel.setTableName(table.getFactTableName)
-    carbonLoadModel.setStorePath(relation.tableMeta.storePath)
-    val dataLoadSchema = new CarbonDataLoadSchema(table)
-    carbonLoadModel.setCarbonDataLoadSchema(dataLoadSchema)
-    try {
-      DataManagementFunc.cleanFiles(
-        sqlContext.sparkContext,
-        carbonLoadModel,
-        relation.tableMeta.storePath)
-      LOGGER.audit(s"Clean files request is successfull for $dbName.$tableName.")
-    } catch {
-      case ex: Exception =>
-        sys.error(ex.getMessage)
-    }
+    Checker.validateTableExists(databaseNameOp, tableName, sqlContext)
+    CarbonStore.cleanFiles(
+      getDB.getDatabaseName(databaseNameOp, sqlContext),
+      tableName,
+      sqlContext.asInstanceOf[CarbonContext].storePath
+    )
     Seq.empty
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
index 62803c7..9cdbf86 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
@@ -148,12 +148,12 @@ class CarbonMetastore(hiveContext: HiveContext, val storePath: String,
                  c.carbonTableIdentifier.getTableName.equalsIgnoreCase(tableName))
   }
 
-  def tableExists(tableIdentifier: TableIdentifier)(sqlContext: SQLContext): Boolean = {
+  def tableExists(identifier: TableIdentifier)(sqlContext: SQLContext): Boolean = {
     checkSchemasModifiedTimeAndReloadTables()
-    val database = tableIdentifier.database.getOrElse(getDB.getDatabaseName(None, sqlContext))
+    val database = identifier.database.getOrElse(getDB.getDatabaseName(None, sqlContext))
     val tables = metadata.tablesMeta.filter(
       c => c.carbonTableIdentifier.getDatabaseName.equalsIgnoreCase(database) &&
-           c.carbonTableIdentifier.getTableName.equalsIgnoreCase(tableIdentifier.table))
+           c.carbonTableIdentifier.getTableName.equalsIgnoreCase(identifier.table))
     tables.nonEmpty
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark/src/test/java/org/carbondata/integration/spark/load/CarbonLoaderUtilTest.java
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/java/org/carbondata/integration/spark/load/CarbonLoaderUtilTest.java b/integration/spark/src/test/java/org/carbondata/integration/spark/load/CarbonLoaderUtilTest.java
index 0161a45..ed4f95b 100644
--- a/integration/spark/src/test/java/org/carbondata/integration/spark/load/CarbonLoaderUtilTest.java
+++ b/integration/spark/src/test/java/org/carbondata/integration/spark/load/CarbonLoaderUtilTest.java
@@ -28,7 +28,6 @@ import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 import org.apache.carbondata.core.carbon.datastore.block.Distributable;
 import org.apache.carbondata.core.carbon.datastore.block.TableBlockInfo;

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 79b1953..de07707 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -431,8 +431,8 @@ object CarbonDataRDDFactory {
                      s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
       }
       // Check if any load need to be deleted before loading new data
-      DataManagementFunc.deleteLoadsAndUpdateMetadata(carbonLoadModel, carbonTable, storePath,
-        isForceDeletion = false)
+      DataManagementFunc.deleteLoadsAndUpdateMetadata(carbonLoadModel.getDatabaseName,
+        carbonLoadModel.getTableName, storePath, isForceDeletion = false)
       if (null == carbonLoadModel.getLoadMetadataDetails) {
         CommonUtil.readLoadMetadataDetails(carbonLoadModel, storePath)
       }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index 10fffd9..68ad4d6 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -17,17 +17,13 @@
 
 package org.apache.spark.sql.execution.command
 
-import java.text.SimpleDateFormat
-
 import scala.collection.JavaConverters._
 import scala.language.implicitConversions
 
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Cast, Literal}
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.hive.{CarbonMetastore, CarbonRelation}
-import org.apache.spark.sql.types.TimestampType
 import org.apache.spark.util.FileUtils
 
 import org.apache.carbondata.common.logging.LogServiceFactory
@@ -40,12 +36,11 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.datastorage.store.impl.FileFactory
 import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
 import org.apache.carbondata.lcm.locks.{CarbonLockFactory, LockUsage}
-import org.apache.carbondata.lcm.status.SegmentStatusManager
 import org.apache.carbondata.processing.constants.TableOptionConstant
 import org.apache.carbondata.processing.etl.DataLoadingException
 import org.apache.carbondata.processing.model.CarbonLoadModel
 import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
-import org.apache.carbondata.spark.rdd.{CarbonDataRDDFactory, DataManagementFunc, DictionaryLoadModel}
+import org.apache.carbondata.spark.rdd.{CarbonDataRDDFactory, DictionaryLoadModel}
 import org.apache.carbondata.spark.util.{CarbonScalaUtil, CarbonSparkUtil, GlobalDictionaryUtil}
 
 /**
@@ -58,7 +53,6 @@ case class AlterTableCompaction(alterTableModel: AlterTableModel) {
   val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
 
   def run(sparkSession: SparkSession): Seq[Row] = {
-    // TODO : Implement it.
     val tableName = alterTableModel.tableName
     val databaseName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
     if (null == org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance
@@ -142,124 +136,6 @@ case class CreateTable(cm: TableModel) {
   }
 }
 
-case class DeleteLoadsById(
-    loadids: Seq[String],
-    databaseNameOp: Option[String],
-    tableName: String) {
-
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  def run(sparkSession: SparkSession): Seq[Row] = {
-
-    val databaseName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
-    LOGGER.audit(s"Delete segment by Id request has been received for $databaseName.$tableName")
-
-    // validate load ids first
-    validateLoadIds
-    val dbName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore.lookupRelation(
-      identifier, None)(sparkSession).asInstanceOf[CarbonRelation]
-    if (relation == null) {
-      LOGGER.audit(s"Delete segment by Id is failed. Table $dbName.$tableName does not exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-
-    val carbonTable = CarbonMetadata.getInstance().getCarbonTable(dbName + '_' + tableName)
-
-    if (null == carbonTable) {
-      CarbonEnv.get.carbonMetastore
-        .lookupRelation(identifier, None)(sparkSession).asInstanceOf[CarbonRelation]
-    }
-    val path = carbonTable.getMetaDataFilepath
-
-    try {
-      val invalidLoadIds = SegmentStatusManager.updateDeletionStatus(
-        carbonTable.getAbsoluteTableIdentifier, loadids.asJava, path).asScala
-
-      if (invalidLoadIds.isEmpty) {
-
-        LOGGER.audit(s"Delete segment by Id is successfull for $databaseName.$tableName.")
-      }
-      else {
-        sys.error("Delete segment by Id is failed. Invalid ID is:" +
-                  s" ${ invalidLoadIds.mkString(",") }")
-      }
-    } catch {
-      case ex: Exception =>
-        sys.error(ex.getMessage)
-    }
-
-    Seq.empty
-
-  }
-
-  // validates load ids
-  private def validateLoadIds: Unit = {
-    if (loadids.isEmpty) {
-      val errorMessage = "Error: Segment id(s) should not be empty."
-      throw new MalformedCarbonCommandException(errorMessage)
-
-    }
-  }
-}
-
-case class DeleteLoadsByLoadDate(
-    databaseNameOp: Option[String],
-    tableName: String,
-    dateField: String,
-    loadDate: String) {
-
-  val LOGGER = LogServiceFactory.getLogService("org.apache.spark.sql.TableModel.tableSchema")
-
-  def run(sparkSession: SparkSession): Seq[Row] = {
-
-    LOGGER.audit("The delete segment by load date request has been received.")
-    val dbName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore
-      .lookupRelation(identifier, None)(sparkSession).asInstanceOf[CarbonRelation]
-    if (relation == null) {
-      LOGGER
-        .audit(s"Delete segment by load date is failed. Table $dbName.$tableName does not " +
-               s"exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-
-    val timeObj = Cast(Literal(loadDate), TimestampType).eval()
-    if (null == timeObj) {
-      val errorMessage = "Error: Invalid load start time format " + loadDate
-      throw new MalformedCarbonCommandException(errorMessage)
-    }
-
-    val carbonTable = org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance()
-      .getCarbonTable(dbName + '_' + tableName)
-
-    if (null == carbonTable) {
-      var relation = CarbonEnv.get.carbonMetastore
-        .lookupRelation(identifier, None)(sparkSession).asInstanceOf[CarbonRelation]
-    }
-    val path = carbonTable.getMetaDataFilepath()
-
-    try {
-      val invalidLoadTimestamps = SegmentStatusManager.updateDeletionStatus(
-        carbonTable.getAbsoluteTableIdentifier, loadDate, path,
-        timeObj.asInstanceOf[java.lang.Long]).asScala
-      if (invalidLoadTimestamps.isEmpty) {
-        LOGGER.audit(s"Delete segment by date is successfull for $dbName.$tableName.")
-      }
-      else {
-        sys.error("Delete segment by date is failed. No matching segment found.")
-      }
-    } catch {
-      case ex: Exception =>
-        sys.error(ex.getMessage)
-    }
-    Seq.empty
-
-  }
-
-}
 
 object LoadTable {
 
@@ -557,152 +433,3 @@ case class LoadTable(
     }
   }
 }
-
-private[sql] case class DeleteLoadByDate(
-    databaseNameOp: Option[String],
-    tableName: String,
-    dateField: String,
-    dateValue: String) {
-
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  def run(sparkSession: SparkSession): Seq[Row] = {
-    val dbName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
-    LOGGER.audit(s"The delete load by date request has been received for $dbName.$tableName")
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore
-      .lookupRelation(identifier)(sparkSession).asInstanceOf[CarbonRelation]
-    var level: String = ""
-    val carbonTable = org.apache.carbondata.core.carbon.metadata.CarbonMetadata
-      .getInstance().getCarbonTable(dbName + '_' + tableName)
-    if (relation == null) {
-      LOGGER.audit(s"The delete load by date is failed. Table $dbName.$tableName does not exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-    val matches: Seq[AttributeReference] = relation.dimensionsAttr.filter(
-      filter => filter.name.equalsIgnoreCase(dateField) &&
-                filter.dataType.isInstanceOf[TimestampType]).toList
-    if (matches.isEmpty) {
-      LOGGER.audit("The delete load by date is failed. " +
-                   s"Table $dbName.$tableName does not contain date field: $dateField")
-      sys.error(s"Table $dbName.$tableName does not contain date field $dateField")
-    } else {
-      level = matches.asJava.get(0).name
-    }
-    val actualColName = relation.metaData.carbonTable.getDimensionByName(tableName, level)
-      .getColName
-    DataManagementFunc.deleteLoadByDate(
-      sparkSession.sqlContext,
-      new CarbonDataLoadSchema(carbonTable),
-      dbName,
-      tableName,
-      CarbonEnv.get.carbonMetastore.storePath,
-      level,
-      actualColName,
-      dateValue)
-    LOGGER.audit(s"The delete load by date $dateValue is successful for $dbName.$tableName.")
-    Seq.empty
-  }
-
-}
-
-case class CleanFiles(
-    databaseNameOp: Option[String],
-    tableName: String) {
-
-  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  def run(sparkSession: SparkSession): Seq[Row] = {
-    val dbName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
-    LOGGER.audit(s"The clean files request has been received for $dbName.$tableName")
-    val identifier = TableIdentifier(tableName, Option(dbName))
-    val relation = CarbonEnv.get.carbonMetastore
-      .lookupRelation(identifier)(sparkSession).asInstanceOf[CarbonRelation]
-    if (relation == null) {
-      LOGGER.audit(s"The clean files request is failed. Table $dbName.$tableName does not exist")
-      sys.error(s"Table $dbName.$tableName does not exist")
-    }
-
-    val carbonLoadModel = new CarbonLoadModel()
-    carbonLoadModel.setTableName(relation.tableMeta.carbonTableIdentifier.getTableName)
-    carbonLoadModel.setDatabaseName(relation.tableMeta.carbonTableIdentifier.getDatabaseName)
-    val table = relation.tableMeta.carbonTable
-    carbonLoadModel.setAggTables(table.getAggregateTablesName.asScala.toArray)
-    carbonLoadModel.setTableName(table.getFactTableName)
-    carbonLoadModel.setStorePath(relation.tableMeta.storePath)
-    val dataLoadSchema = new CarbonDataLoadSchema(table)
-    carbonLoadModel.setCarbonDataLoadSchema(dataLoadSchema)
-    try {
-      DataManagementFunc.cleanFiles(
-        sparkSession.sqlContext.sparkContext,
-        carbonLoadModel,
-        relation.tableMeta.storePath)
-      LOGGER.audit(s"Clean files request is successfull for $dbName.$tableName.")
-    } catch {
-      case ex: Exception =>
-        sys.error(ex.getMessage)
-    }
-    Seq.empty
-  }
-}
-
-case class ShowLoads(
-    databaseNameOp: Option[String],
-    tableName: String,
-    limit: Option[String],
-    val output: Seq[Attribute]) {
-
-  def run(sparkSession: SparkSession): Seq[Row] = {
-    val databaseName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
-    val tableUniqueName = databaseName + "_" + tableName
-    // Here using checkSchemasModifiedTimeAndReloadTables in tableExists to reload metadata if
-    // schema is changed by other process, so that tableInfoMap woulb be refilled.
-    val tableExists = CarbonEnv.get.carbonMetastore
-        .tableExists(TableIdentifier(tableName, databaseNameOp))(sparkSession)
-    if (!tableExists) {
-      sys.error(s"$databaseName.$tableName is not found")
-    }
-    val carbonTable = org.apache.carbondata.core.carbon.metadata.CarbonMetadata.getInstance()
-        .getCarbonTable(tableUniqueName)
-    if (carbonTable == null) {
-      sys.error(s"$databaseName.$tableName is not found")
-    }
-    val path = carbonTable.getMetaDataFilepath
-    val loadMetadataDetailsArray = SegmentStatusManager.readLoadMetadata(path)
-    if (loadMetadataDetailsArray.nonEmpty) {
-
-      val parser = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP)
-
-      var loadMetadataDetailsSortedArray = loadMetadataDetailsArray.sortWith(
-        (l1, l2) => java.lang.Double.parseDouble(l1.getLoadName) > java.lang.Double
-            .parseDouble(l2.getLoadName)
-      )
-
-
-      if (limit.isDefined) {
-        loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray
-            .filter(load => load.getVisibility.equalsIgnoreCase("true"))
-        val limitLoads = limit.get
-        try {
-          val lim = Integer.parseInt(limitLoads)
-          loadMetadataDetailsSortedArray = loadMetadataDetailsSortedArray.slice(0, lim)
-        } catch {
-          case ex: NumberFormatException => sys.error(s" Entered limit is not a valid Number")
-        }
-
-      }
-
-      loadMetadataDetailsSortedArray.filter(load => load.getVisibility.equalsIgnoreCase("true"))
-          .map(load =>
-            Row(
-              load.getLoadName,
-              load.getLoadStatus,
-              new java.sql.Timestamp(parser.parse(load.getLoadStartTime).getTime),
-              new java.sql.Timestamp(parser.parse(load.getTimestamp).getTime))).toSeq
-    } else {
-      Seq.empty
-
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
index 9467804..9638b8f 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
@@ -176,12 +176,14 @@ class CarbonMetastore(conf: RuntimeConfig, val storePath: String) {
                  c.carbonTableIdentifier.getTableName.equalsIgnoreCase(tableName))
   }
 
-  def tableExists(tableIdentifier: TableIdentifier)(sparkSession: SparkSession): Boolean = {
+  def tableExists(
+      table: String,
+      databaseOp: Option[String] = None)(sparkSession: SparkSession): Boolean = {
     checkSchemasModifiedTimeAndReloadTables()
-    val database = tableIdentifier.database.getOrElse(sparkSession.catalog.currentDatabase)
+    val database = databaseOp.getOrElse(sparkSession.catalog.currentDatabase)
     val tables = metadata.tablesMeta.filter(
       c => c.carbonTableIdentifier.getDatabaseName.equalsIgnoreCase(database) &&
-           c.carbonTableIdentifier.getTableName.equalsIgnoreCase(tableIdentifier.table))
+           c.carbonTableIdentifier.getTableName.equalsIgnoreCase(table))
     tables.nonEmpty
   }
 
@@ -294,7 +296,7 @@ class CarbonMetastore(conf: RuntimeConfig, val storePath: String) {
       tableInfo: org.apache.carbondata.core.carbon.metadata.schema.table.TableInfo,
       dbName: String, tableName: String)
     (sparkSession: SparkSession): String = {
-    if (tableExists(TableIdentifier(tableName, Some(dbName)))(sparkSession)) {
+    if (tableExists(tableName, Some(dbName))(sparkSession)) {
       sys.error(s"Table [$tableName] already exists under Database [$dbName]")
     }
     val schemaConverter = new ThriftWrapperSchemaConverterImpl

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/util/CleanFiles.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/CleanFiles.scala b/integration/spark2/src/main/scala/org/apache/spark/util/CleanFiles.scala
index f493af1..c84882e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/CleanFiles.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/CleanFiles.scala
@@ -18,7 +18,8 @@
 package org.apache.spark.util
 
 import org.apache.spark.sql.{CarbonEnv, SparkSession}
-import org.apache.spark.sql.execution.command.{CleanFiles => TableCleanFiles}
+
+import org.apache.carbondata.api.CarbonStore
 
 /**
  * clean files api
@@ -26,21 +27,23 @@ import org.apache.spark.sql.execution.command.{CleanFiles => TableCleanFiles}
  // scalastyle:off
 object CleanFiles {
 
-  def cleanFiles(spark: SparkSession, dbName: Option[String], tableName: String): Unit = {
-    TableCleanFiles(dbName, tableName).run(spark)
+  def cleanFiles(spark: SparkSession, dbName: String, tableName: String,
+      storePath: String): Unit = {
+    TableAPIUtil.validateTableExists(spark, dbName, tableName)
+    CarbonStore.cleanFiles(dbName, tableName, storePath)
   }
 
   def main(args: Array[String]): Unit = {
 
     if (args.length < 2) {
-      System.err.println("Usage: TableCleanFiles <store path> <table name>");
+      System.err.println("Usage: CleanFiles <store path> <table name>");
       System.exit(1)
     }
 
     val storePath = TableAPIUtil.escape(args(0))
     val (dbName, tableName) = TableAPIUtil.parseSchemaName(TableAPIUtil.escape(args(1)))
-    val spark = TableAPIUtil.spark(storePath, s"TableCleanFiles: $dbName.$tableName")
+    val spark = TableAPIUtil.spark(storePath, s"CleanFiles: $dbName.$tableName")
     CarbonEnv.init(spark.sqlContext)
-    cleanFiles(spark, Option(dbName), tableName)
+    cleanFiles(spark, dbName, tableName, storePath)
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/util/Compaction.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/Compaction.scala b/integration/spark2/src/main/scala/org/apache/spark/util/Compaction.scala
index 90310d3..1e891fd 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/Compaction.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/Compaction.scala
@@ -25,22 +25,23 @@ import org.apache.spark.sql.execution.command.{AlterTableCompaction, AlterTableM
  // scalastyle:off
 object Compaction {
 
-  def compaction(spark: SparkSession, dbName: Option[String], tableName: String,
+  def compaction(spark: SparkSession, dbName: String, tableName: String,
       compactionType: String): Unit = {
-    AlterTableCompaction(AlterTableModel(dbName, tableName, compactionType, "")).run(spark)
+    TableAPIUtil.validateTableExists(spark, dbName, tableName)
+    AlterTableCompaction(AlterTableModel(Some(dbName), tableName, compactionType, "")).run(spark)
   }
 
   def main(args: Array[String]): Unit = {
     if (args.length < 3) {
-      System.err.println("Usage: TableCompaction <store path> <table name> <major|minor>");
+      System.err.println("Usage: Compaction <store path> <table name> <major|minor>");
       System.exit(1)
     }
 
     val storePath = TableAPIUtil.escape(args(0))
     val (dbName, tableName) = TableAPIUtil.parseSchemaName(TableAPIUtil.escape(args(1)))
     val compactionType = TableAPIUtil.escape(args(2))
-    val spark = TableAPIUtil.spark(storePath, s"TableCompaction: $dbName.$tableName")
+    val spark = TableAPIUtil.spark(storePath, s"Compaction: $dbName.$tableName")
     CarbonEnv.init(spark.sqlContext)
-    compaction(spark, Option(dbName), tableName, compactionType)
+    compaction(spark, dbName, tableName, compactionType)
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentByDate.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentByDate.scala b/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentByDate.scala
index f77a16e..ae95bf6 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentByDate.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentByDate.scala
@@ -17,7 +17,8 @@
  package org.apache.spark.util
 
 import org.apache.spark.sql.{CarbonEnv, SparkSession}
-import org.apache.spark.sql.execution.command.DeleteLoadsByLoadDate
+
+import org.apache.carbondata.api.CarbonStore
 
 /**
  * delete segments before some date
@@ -25,23 +26,24 @@ import org.apache.spark.sql.execution.command.DeleteLoadsByLoadDate
 // scalastyle:off
 object DeleteSegmentByDate {
 
-  def deleteSegmentByDate(spark: SparkSession, dbName: Option[String], tableName: String,
+  def deleteSegmentByDate(spark: SparkSession, dbName: String, tableName: String,
       dateValue: String): Unit = {
-    DeleteLoadsByLoadDate(dbName, tableName, "", dateValue).run(spark)
+    TableAPIUtil.validateTableExists(spark, dbName, tableName)
+    CarbonStore.deleteLoadByDate(dateValue, dbName, tableName)
   }
 
   def main(args: Array[String]): Unit = {
     if (args.length < 3) {
       System.err.println(
-        "Usage: TableDeleteSegmentByDate <store path> <table name> <before date value>");
+        "Usage: DeleteSegmentByDate <store path> <table name> <before date value>");
       System.exit(1)
     }
 
     val storePath = TableAPIUtil.escape(args(0))
     val (dbName, tableName) = TableAPIUtil.parseSchemaName(TableAPIUtil.escape(args(1)))
     val dateValue = TableAPIUtil.escape(args(2))
-    val spark = TableAPIUtil.spark(storePath, s"TableCleanFiles: $dbName.$tableName")
+    val spark = TableAPIUtil.spark(storePath, s"DeleteSegmentByDate: $dbName.$tableName")
     CarbonEnv.init(spark.sqlContext)
-    deleteSegmentByDate(spark, Option(dbName), tableName, dateValue)
+    deleteSegmentByDate(spark, dbName, tableName, dateValue)
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentById.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentById.scala b/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentById.scala
index c3e8626..d5a6861 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentById.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/DeleteSegmentById.scala
@@ -17,7 +17,8 @@
 package org.apache.spark.util
 
 import org.apache.spark.sql.{CarbonEnv, SparkSession}
-import org.apache.spark.sql.execution.command.DeleteLoadsById
+
+import org.apache.carbondata.api.CarbonStore
 
 /**
  * delete segments by id list
@@ -29,24 +30,25 @@ object DeleteSegmentById {
     segmentIds.split(",").toSeq
   }
 
-  def deleteSegmentById(spark: SparkSession, dbName: Option[String], tableName: String,
+  def deleteSegmentById(spark: SparkSession, dbName: String, tableName: String,
       segmentIds: Seq[String]): Unit = {
-    DeleteLoadsById(segmentIds, dbName, tableName).run(spark)
+    TableAPIUtil.validateTableExists(spark, dbName, tableName)
+    CarbonStore.deleteLoadById(segmentIds, dbName, tableName)
   }
 
   def main(args: Array[String]): Unit = {
 
     if (args.length < 3) {
       System.err.println(
-        "Usage: TableDeleteSegmentByID <store path> <table name> <segment id list>");
+        "Usage: DeleteSegmentByID <store path> <table name> <segment id list>");
       System.exit(1)
     }
 
     val storePath = TableAPIUtil.escape(args(0))
     val (dbName, tableName) = TableAPIUtil.parseSchemaName(TableAPIUtil.escape(args(1)))
     val segmentIds = extractSegmentIds(TableAPIUtil.escape(args(2)))
-    val spark = TableAPIUtil.spark(storePath, s"TableDeleteSegmentById: $dbName.$tableName")
+    val spark = TableAPIUtil.spark(storePath, s"DeleteSegmentById: $dbName.$tableName")
     CarbonEnv.init(spark.sqlContext)
-    deleteSegmentById(spark, Option(dbName), tableName, segmentIds)
+    deleteSegmentById(spark, dbName, tableName, segmentIds)
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/util/ShowSegments.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/ShowSegments.scala b/integration/spark2/src/main/scala/org/apache/spark/util/ShowSegments.scala
index 5ddffcd..1a02c8c 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/ShowSegments.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/ShowSegments.scala
@@ -17,41 +17,41 @@
 
 package org.apache.spark.util
 
+import java.text.SimpleDateFormat
+
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.execution.command.ShowLoads
-import org.apache.spark.sql.types.{StringType, TimestampType}
+
+import org.apache.carbondata.api.CarbonStore
 
 // scalastyle:off
 object ShowSegments {
 
-  def showSegments(spark: SparkSession, dbName: Option[String], tableName: String,
+  def showSegments(spark: SparkSession, dbName: String, tableName: String,
       limit: Option[String]): Seq[Row] = {
-    val output =  Seq(AttributeReference("SegmentSequenceId", StringType, nullable = false)(),
-      AttributeReference("Status", StringType, nullable = false)(),
-      AttributeReference("Load Start Time", TimestampType, nullable = false)(),
-      AttributeReference("Load End Time", TimestampType, nullable = false)())
-    ShowLoads(dbName, tableName, limit: Option[String], output).run(spark)
+    //val databaseName = dbName.getOrElse(spark.catalog.currentDatabase)
+    TableAPIUtil.validateTableExists(spark, dbName, tableName)
+    CarbonStore.showSegments(dbName, tableName, limit)
   }
 
   def showString(rows: Seq[Row]): String = {
+    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.s")
     val sb = new StringBuilder
-    sb.append("+-----------------+---------------+---------------------+---------------------+\n")
-      .append("|SegmentSequenceId|Status         |Load Start Time      |Load End Time        |\n")
-      .append("+-----------------+---------------+---------------------+---------------------+\n")
+    sb.append("+------------+------------------+----------------------+----------------------+\n")
+      .append("|SegmentId   |Status            |Load Start Time       |Load End Time         |\n")
+      .append("+------------+------------------+----------------------+----------------------+\n")
       rows.foreach{row =>
         sb.append("|")
-          .append(StringUtils.rightPad(row.getString(0), 17))
+          .append(StringUtils.rightPad(row.getString(0), 12))
           .append("|")
-          .append(StringUtils.rightPad(row.getString(1).substring(0, 15), 15))
+          .append(StringUtils.rightPad(row.getString(1), 18))
           .append("|")
-          .append(row.getAs[java.sql.Timestamp](2).formatted("yyyy-MM-dd HH:mm:ss.s"))
+          .append(sdf.format(row.getAs[java.sql.Timestamp](2)))
           .append("|")
-          .append(row.getAs[java.sql.Timestamp](3).formatted("yyyy-MM-dd HH:mm:ss.s"))
+          .append(sdf.format(row.getAs[java.sql.Timestamp](3)))
           .append("|\n")
       }
-    sb.append("+-----------------+---------------+---------------------+---------------------+\n")
+    sb.append("+------------+------------------+----------------------+----------------------+\n")
     sb.toString
   }
 
@@ -74,9 +74,9 @@ object ShowSegments {
     } else {
       None
     }
-    val spark = TableAPIUtil.spark(storePath, s"TableCleanFiles: $dbName.$tableName")
+    val spark = TableAPIUtil.spark(storePath, s"ShowSegments: $dbName.$tableName")
     CarbonEnv.init(spark.sqlContext)
-    val rows = showSegments(spark, Option(dbName), tableName, limit)
+    val rows = showSegments(spark, dbName, tableName, limit)
     System.out.println(showString(rows))
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/b50866b3/integration/spark2/src/main/scala/org/apache/spark/util/TableAPIUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/TableAPIUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/util/TableAPIUtil.scala
index 6954981..cb444d6 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/TableAPIUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/TableAPIUtil.scala
@@ -17,14 +17,20 @@
 
 package org.apache.spark.util
 
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{CarbonEnv, SparkSession}
 
+import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
 
 /**
  * table api util
  */
 object TableAPIUtil {
+
+  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+
   def parseSchemaName(tableName: String): (String, String) = {
     if (tableName.contains(".")) {
       val parts = tableName.split(".")
@@ -44,11 +50,22 @@ object TableAPIUtil {
   }
 
   def spark(storePath: String, appName: String): SparkSession = {
+    // CarbonEnv depends on CarbonProperty to get the store path, so set it here
+    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)
     SparkSession
         .builder
         .appName(appName)
-        .master("local")
-        .config(CarbonCommonConstants.STORE_LOCATION, storePath)
         .getOrCreate()
   }
+
+  def validateTableExists(
+      spark: SparkSession,
+      dbName: String,
+      tableName: String): Unit = {
+    if (!CarbonEnv.get.carbonMetastore.tableExists(tableName, Some(dbName))(spark)) {
+      val err = s"table $dbName.$tableName not found"
+      LOGGER.error(err)
+      throw new MalformedCarbonCommandException(err)
+    }
+  }
 }



Mime
View raw message