fixed issue with meta lock not getting deleted for rename table

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

@@ -18,6 +18,7 @@
 package org.apache.carbondata.core.locks;
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
 import org.apache.carbondata.core.util.CarbonProperties;
 
 /**
@@ -72,4 +73,12 @@ protected void initRetry() {
 
   }
 
+  public boolean releaseLockManually(String lockFile) {
+    try {
+      return FileFactory.deleteFile(lockFile, FileFactory.getFileType(lockFile));
+    } catch (Exception e) {
+      return false;
+    }
+  }
+
 }

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

@@ -23,6 +23,7 @@
 import org.apache.carbondata.common.logging.LogService;
 import org.apache.carbondata.common.logging.LogServiceFactory;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
 import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
 import org.apache.carbondata.core.util.CarbonProperties;
@@ -102,34 +103,31 @@ public HdfsFileLock(CarbonTableIdentifier tableIdentifier, String lockFile) {
    */
   @Override
   public boolean unlock() {
+    boolean status = false;
     if (null != dataOutputStream) {
       try {
         dataOutputStream.close();
+        status = true;
       } catch (IOException e) {
-        try {
-          if (!FileFactory.isFileExist(location, FileFactory.getFileType(location))) {
-            return true;
-          }
-        } catch (IOException e1) {
-          LOGGER.error("Exception in isFileExist of the lock file " + e1.getMessage());
-        }
-        LOGGER.error("Exception in unlocking of the lock file " + e.getMessage());
-        return false;
+        status = false;
       } finally {
-        try {
-          if (FileFactory.isFileExist(location, FileFactory.getFileType(location))) {
-            if (FileFactory.getCarbonFile(location, FileFactory.getFileType(location)).delete()) {
-              LOGGER.info("Deleted the lock file " + location);
-            } else {
-              LOGGER.error("Not able to delete the lock file " + location);
-            }
+        CarbonFile carbonFile =
+            FileFactory.getCarbonFile(location, FileFactory.getFileType(location));
+        if (carbonFile.exists()) {
+          if (carbonFile.delete()) {
+            LOGGER.info("Deleted the lock file " + location);
+          } else {
+            LOGGER.error("Not able to delete the lock file " + location);
+            status = false;
           }
-        } catch (IOException e) {
-          LOGGER.error("Exception in isFileExist of the lock file " + e.getMessage());
+        } else {
+          LOGGER.error("Not able to delete the lock file because "
+              + "it is not existed in location " + location);
+          status = false;
         }
       }
     }
-    return true;
+    return status;
   }
 
 }

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

@@ -36,4 +36,12 @@ public interface ICarbonLock {
    */
   boolean lockWithRetries();
 
+  /**
+   * This method will delete the lock file at the specified location.
+   *
+   * @param lockFile The path of the lock file.
+   * @return True if the lock file is deleted, false otherwise.
+   */
+  boolean releaseLockManually(String lockFile);
+
 }

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

@@ -28,6 +28,7 @@
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
 import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
+import org.apache.carbondata.core.util.CarbonProperties;
 
 /**
  * This class handles the file locking in the local file system.
@@ -70,7 +71,8 @@ public class LocalFileLock extends AbstractCarbonLock {
       LogServiceFactory.getLogService(LocalFileLock.class.getName());
 
   static {
-    tmpPath = System.getProperty("java.io.tmpdir");
+    tmpPath = CarbonProperties.getInstance().getProperty(CarbonCommonConstants.STORE_LOCATION,
+        System.getProperty("java.io.tmpdir"));
   }
 
   /**
@@ -151,6 +153,7 @@ public LocalFileLock(CarbonTableIdentifier tableIdentifier, String lockFile) {
             LOGGER.info("Successfully deleted the lock file " + lockFilePath);
           } else {
             LOGGER.error("Not able to delete the lock file " + lockFilePath);
+            status = false;
           }
         } catch (IOException e) {
           LOGGER.error(e.getMessage());

integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/AlterTableCommands.scala

@@ -33,7 +33,7 @@ import org.apache.carbondata.core.metadata.{CarbonMetadata, CarbonTableIdentifie
 import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
 import org.apache.carbondata.core.metadata.encoder.Encoding
 import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
-import org.apache.carbondata.core.util.CarbonUtil
+import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
 import org.apache.carbondata.core.util.path.CarbonStorePath
 import org.apache.carbondata.format.{ColumnSchema, SchemaEvolutionEntry, TableInfo}
 import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
@@ -204,7 +204,18 @@ private[sql] case class AlterTableRenameTable(alterTableRenameModel: AlterTableR
         if (carbonLock.unlock()) {
           LOGGER.info("Lock released successfully")
         } else {
-          LOGGER.error("Unable to release lock during rename table")
+          val storeLocation = CarbonProperties.getInstance
+            .getProperty(CarbonCommonConstants.STORE_LOCATION,
+              System.getProperty("java.io.tmpdir"))
+          val lockFilePath = storeLocation + CarbonCommonConstants.FILE_SEPARATOR +
+                             oldDatabaseName + CarbonCommonConstants.FILE_SEPARATOR + newTableName +
+                             CarbonCommonConstants.FILE_SEPARATOR +
+                             LockUsage.METADATA_LOCK
+          if(carbonLock.releaseLockManually(lockFilePath)) {
+            LOGGER.info("Lock released successfully")
+          } else {
+            LOGGER.error("Unable to release lock during rename table")
+          }
         }
       }
     }

integration/spark2/src/main/scala/org/apache/spark/util/AlterTableUtil.scala

@@ -30,6 +30,7 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable
 import org.apache.carbondata.format.{SchemaEvolutionEntry, TableInfo}
 
 object AlterTableUtil {
+
   def validateTableAndAcquireLock(dbName: String, tableName: String, LOGGER: LogService)
     (sparkSession: SparkSession): ICarbonLock = {
     val relation =

integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala

@@ -381,11 +381,19 @@ class AlterTableValidationTestCase extends QueryTest with BeforeAndAfterAll {
     sql("drop database testdb")
   }
 
+  test("test to check if the lock file is successfully deleted") {
+    sql("create table lock_check(id int, name string) stored by 'carbondata'")
+    sql("alter table lock_check rename to lock_rename")
+    assert(!new File(s"${ CarbonCommonConstants.STORE_LOCATION } + /default/lock_rename/meta.lock")
+      .exists())
+  }
+
   override def afterAll {
     sql("DROP TABLE IF EXISTS restructure")
     sql("DROP TABLE IF EXISTS restructure_new")
     sql("DROP TABLE IF EXISTS restructure_test")
     sql("DROP TABLE IF EXISTS restructure_bad")
     sql("DROP TABLE IF EXISTS restructure_badnew")
+    sql("DROP TABLE IF EXISTS lock_rename")
   }
 }