carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From manishgupt...@apache.org
Subject carbondata git commit: [CARBONDATA-1876] Clean all the InProgress segments for all databases during session initialization
Date Wed, 13 Dec 2017 12:42:19 GMT
Repository: carbondata
Updated Branches:
  refs/heads/master 4158f3d96 -> 6a185b834


[CARBONDATA-1876] Clean all the InProgress segments for all databases during session initialization

Clean all the InProgress segments for all databases during session initialization. when carbon
session initialize, clean all the in progress segments for all the databases created by user.

This closes #1637


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

Branch: refs/heads/master
Commit: 6a185b8341f6f36f21b325c961a0a33ab17b15aa
Parents: 4158f3d
Author: akashrn5 <akashnilugal@gmail.com>
Authored: Fri Dec 8 04:36:55 2017 +0530
Committer: manishgupta88 <tomanishgupta18@gmail.com>
Committed: Wed Dec 13 18:15:11 2017 +0530

----------------------------------------------------------------------
 .../core/constants/CarbonCommonConstants.java   |  2 +-
 .../carbondata/spark/util/CommonUtil.scala      | 24 +++++++++-----------
 .../org/apache/spark/sql/CarbonSession.scala    |  7 ++++--
 3 files changed, 17 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/6a185b83/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index 4ab9919..959f4d4 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -1302,7 +1302,7 @@ public final class CarbonCommonConstants {
    */
   public static final String DATA_MANAGEMENT_DRIVER = "spark.carbon.datamanagement.driver";
 
-  public static final String DATA_MANAGEMENT_DRIVER_DEFAULT = "false";
+  public static final String DATA_MANAGEMENT_DRIVER_DEFAULT = "true";
 
   public static final String CARBON_SESSIONSTATE_CLASSNAME = "spark.carbon.sessionstate.classname";
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6a185b83/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index 3595884..943c0a5 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -760,10 +760,10 @@ object CommonUtil {
   /**
    * The in-progress segments which are left when the driver is down will be marked as deleted
    * when driver is initializing.
-   * @param storePath
-   * @param sparkContext
+   * @param databaseLocation
+   * @param dbName
    */
-  def cleanInProgressSegments(storePath: String, sparkContext: SparkContext): Unit = {
+  def cleanInProgressSegments(databaseLocation: String, dbName: String): Unit = {
     val loaderDriver = CarbonProperties.getInstance().
       getProperty(CarbonCommonConstants.DATA_MANAGEMENT_DRIVER,
         CarbonCommonConstants.DATA_MANAGEMENT_DRIVER_DEFAULT).toBoolean
@@ -771,18 +771,17 @@ object CommonUtil {
       return
     }
     try {
-      val fileType = FileFactory.getFileType(storePath)
-      if (FileFactory.isFileExist(storePath, fileType)) {
-        val file = FileFactory.getCarbonFile(storePath, fileType)
-        val databaseFolders = file.listFiles()
-        databaseFolders.foreach { databaseFolder =>
-          if (databaseFolder.isDirectory) {
-            val tableFolders = databaseFolder.listFiles()
+      val fileType = FileFactory.getFileType(databaseLocation)
+      if (FileFactory.isFileExist(databaseLocation, fileType)) {
+        val file = FileFactory.getCarbonFile(databaseLocation, fileType)
+          if (file.isDirectory) {
+            val tableFolders = file.listFiles()
             tableFolders.foreach { tableFolder =>
               if (tableFolder.isDirectory) {
+                val tablePath = databaseLocation +
+                                CarbonCommonConstants.FILE_SEPARATOR + tableFolder.getName
                 val identifier =
-                  AbsoluteTableIdentifier.from(storePath,
-                    databaseFolder.getName, tableFolder.getName)
+                  AbsoluteTableIdentifier.from(tablePath, dbName, tableFolder.getName)
                 val carbonTablePath = CarbonStorePath.getCarbonTablePath(identifier)
                 val tableStatusFile = carbonTablePath.getTableStatusFilePath
                 if (FileFactory.isFileExist(tableStatusFile, fileType)) {
@@ -826,7 +825,6 @@ object CommonUtil {
               }
             }
           }
-        }
       }
     } catch {
       case s: java.io.FileNotFoundException =>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6a185b83/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
index e6ee535..c9b134e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
@@ -182,8 +182,11 @@ object CarbonSession {
         options.foreach { case (k, v) => session.sessionState.conf.setConfString(k, v)
}
         SparkSession.setDefaultSession(session)
         try {
-          CommonUtil.cleanInProgressSegments(
-            carbonProperties.getProperty(CarbonCommonConstants.STORE_LOCATION), sparkContext)
+          val databases = session.sessionState.catalog.listDatabases()
+          databases.foreach(dbName => {
+            val databaseLocation = CarbonEnv.getDatabaseLocation(dbName, session)
+            CommonUtil.cleanInProgressSegments(databaseLocation, dbName)
+          })
         } catch {
           case e: Throwable =>
             // catch all exceptions to avoid CarbonSession initialization failure


Mime
View raw message