code clean

Author: marchpure

core/src/main/java/org/apache/carbondata/core/locks/CarbonLockUtil.java

@@ -17,6 +17,9 @@
 
 package org.apache.carbondata.core.locks;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.carbondata.common.logging.LogServiceFactory;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
@@ -134,7 +137,6 @@ public static void deleteExpiredSegmentLockFiles(CarbonTable carbonTable) {
     }
     CarbonFile[] files = FileFactory.getCarbonFile(lockFilesDir)
         .listFiles(new CarbonFileFilter() {
-
             @Override
             public boolean accept(CarbonFile pathName) {
               if (CarbonTablePath.isSegmentLockFilePath(pathName.getName())) {
@@ -149,4 +151,28 @@ public boolean accept(CarbonFile pathName) {
       file.delete();
     }
   }
+
+  public static List<ICarbonLock> acquireLock(CarbonTable carbonTable,
+      List<String> locksToBeAcquired) {
+    List<ICarbonLock> acquiredLocks = new ArrayList<>();
+    try {
+      locksToBeAcquired.forEach(lock ->
+          acquiredLocks.add(CarbonLockUtil
+              .getLockObject(carbonTable.getAbsoluteTableIdentifier(), lock))
+      );
+    } catch (Exception e) {
+      releaseLocks(acquiredLocks);
+    }
+    return acquiredLocks;
+  }
+
+  public static void releaseLocks(List<ICarbonLock> locks) {
+    locks.forEach(carbonLock -> {
+      if (carbonLock.unlock()) {
+        LOGGER.info("Alter table lock released successfully");
+      } else {
+        LOGGER.error("Unable to release lock");
+      }
+    });
+  }
 }

core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java

@@ -176,7 +176,6 @@ public static String getDeleteDeltaFilePath(String blockPath, String blockName,
                                               String timestamp) {
     return blockPath + CarbonCommonConstants.FILE_SEPARATOR + blockName
         + CarbonCommonConstants.HYPHEN + timestamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT;
-
   }
 
   /**
@@ -280,7 +279,7 @@ public static void mergeSegmentUpdate(boolean isCompaction, List<SegmentUpdateDe
   public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
       CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired,
       boolean isUpdateStatusFileUpdateRequired, List<Segment> segmentsToBeDeleted,
-      LoadMetadataDetails newLoadEntry) throws IOException{
+      LoadMetadataDetails newLoadEntry) throws IOException {
     return updateTableMetadataStatus(updatedSegmentsList, table, updatedTimeStamp,
         isTimestampUpdateRequired, isUpdateStatusFileUpdateRequired,
         segmentsToBeDeleted, new ArrayList<Segment>(), "", newLoadEntry);
@@ -298,7 +297,8 @@ public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList
   public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
       CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired,
       boolean isUpdateStatusFileUpdateRequired, List<Segment> segmentsToBeDeleted,
-      List<Segment> segmentFilesTobeUpdated, String uuid, LoadMetadataDetails newLoadEntry) throws IOException{
+      List<Segment> segmentFilesTobeUpdated, String uuid,
+      LoadMetadataDetails newLoadEntry) throws IOException {
 
     boolean status = false;
     String metaDataFilepath = table.getMetadataPath();
@@ -370,8 +370,7 @@ public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList
             indexToOverwriteNewMetaEntry++;
           }
           if (!found) {
-            LOGGER.error("Entry not found to update " + newLoadEntry + " From list :: "
-                    + listOfLoadFolderDetailsArray);
+            LOGGER.error("Entry not found to update " + newLoadEntry + " From list");
             throw new IOException("Entry not found to update in the table status file");
           }
           listOfLoadFolderDetailsArray[indexToOverwriteNewMetaEntry] = newLoadEntry;
@@ -814,7 +813,7 @@ public static long readCurrentTime() {
    * @param segmentBlockCount
    */
   public static void decrementDeletedBlockCount(SegmentUpdateDetails details,
-                                                Map<String, Long> segmentBlockCount) {
+      Map<String, Long> segmentBlockCount) {
 
     String segId = details.getSegmentName();
 

hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java

@@ -19,7 +19,6 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -35,7 +34,6 @@
 import org.apache.carbondata.core.locks.LockUsage;
 import org.apache.carbondata.core.metadata.SegmentFileStore;
 import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
-import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
 import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
 import org.apache.carbondata.core.statusmanager.SegmentStatus;
 import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
@@ -287,13 +285,16 @@ private void commitJobForPartition(JobContext context, boolean overwriteSet,
       newMetaEntry.setDataSize(size);
     }
     String uniqueId = null;
-    String updateTime = context.getConfiguration().get(CarbonTableOutputFormat.UPDATE_TIMESTAMP, uniqueId);
+    String updateTime = context.getConfiguration()
+        .get(CarbonTableOutputFormat.UPDATE_TIMESTAMP, uniqueId);
     if (overwriteSet) {
       uniqueId = overwritePartitions(loadModel, newMetaEntry, uuid);
-    } else if (StringUtils.isNotEmpty(updateTime)){
-      context.getConfiguration().set("carbon.newMetaEntry", ObjectSerializationUtil.convertObjectToString(newMetaEntry));
+    } else if (StringUtils.isNotEmpty(updateTime)) {
+      context.getConfiguration().set("carbon.newMetaEntry",
+          ObjectSerializationUtil.convertObjectToString(newMetaEntry));
     } else {
-      CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid);
+      CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry,
+          loadModel, false, false, uuid);
     }
     if (operationContext != null) {
       operationContext

integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAddLoadCommand.scala

@@ -227,7 +227,6 @@ case class CarbonAddLoadCommand(
     model.setDatabaseName(carbonTable.getDatabaseName)
     model.setTableName(carbonTable.getTableName)
     val operationContext = new OperationContext
-    operationContext.setProperty("isLoadOrCompaction", false)
     val (tableIndexes, indexOperationContext) = CommonLoadUtils.firePreLoadEvents(sparkSession,
       model,
       "",

integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoHadoopFsRelationCommand.scala

@@ -19,10 +19,8 @@ package org.apache.spark.sql.execution.command.management
 
 import java.io.IOException
 
-import org.apache.carbondata.core.statusmanager.LoadMetadataDetails
-import org.apache.carbondata.core.util.ObjectSerializationUtil
-
 import scala.collection.mutable
+
 import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.internal.io.FileCommitProtocol
 import org.apache.spark.sql._
@@ -34,9 +32,10 @@ import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.execution.command._
 import org.apache.spark.sql.execution.datasources.{CarbonSQLHadoopMapReduceCommitProtocol, FileFormat, FileFormatWriter, FileIndex, PartitioningUtils, SparkCarbonTableFormat}
 import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
+import org.apache.spark.sql.types.StringType
 import org.apache.spark.sql.util.SchemaUtils
+
 import org.apache.carbondata.spark.util.CarbonScalaUtil
-import org.apache.spark.sql.types.StringType
 
 /**
  * A command for writing data to a

integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoWithDf.scala

@@ -24,12 +24,12 @@ import scala.collection.JavaConverters._
 import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SparkSession}
 import org.apache.spark.sql.execution.command.UpdateTableModel
 import org.apache.spark.util.CausedBy
+
 import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.datastore.impl.FileFactory
 import org.apache.carbondata.core.metadata.schema.table.TableInfo
 import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
-import org.apache.carbondata.core.util.ObjectSerializationUtil.convertStringToObject
-import org.apache.carbondata.core.util.{CarbonProperties, ObjectSerializationUtil, ThreadLocalSessionInfo}
+import org.apache.carbondata.core.util.{CarbonProperties, ThreadLocalSessionInfo}
 import org.apache.carbondata.core.util.path.CarbonTablePath
 import org.apache.carbondata.events.OperationContext
 import org.apache.carbondata.events.exception.PreEventException

integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CommonLoadUtils.scala

@@ -23,6 +23,7 @@ import java.util
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
+
 import org.apache.commons.lang3.StringUtils
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.rdd.RDD
@@ -41,6 +42,7 @@ import org.apache.spark.sql.util.SparkSQLUtil
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.{CarbonReflectionUtils, CollectionAccumulator, SparkUtil}
+
 import org.apache.carbondata.common.{Maps, Strings}
 import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.converter.SparkDataTypeConverterImpl
@@ -1051,11 +1053,11 @@ object CommonLoadUtils {
       ds.collect()
 
       if (!loadParams.updateModel.isEmpty && loadParams.updateModel.isDefined
-        && table.isHivePartitionTable) {
+          && table.isHivePartitionTable
+          && !ds.logicalPlan.asInstanceOf[LocalRelation].data(0).anyNull) {
         loadParams.updateModel.get.addedLoadDetail =
           Some(ObjectSerializationUtil.convertStringToObject(
-            ds.logicalPlan
-              .asInstanceOf[LocalRelation].data(0).getString(0))
+            ds.logicalPlan.asInstanceOf[LocalRelation].data(0).getString(0))
             .asInstanceOf[LoadMetadataDetails])
       }
     } catch {