carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [19/50] [abbrv] carbondata git commit: [CARBONDATA-2717] fixed table id empty problem while taking drop lock
Date Wed, 18 Jul 2018 02:19:52 GMT
[CARBONDATA-2717] fixed table id empty problem while taking drop lock

This closes #2472


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

Branch: refs/heads/carbonstore
Commit: 637a97469c1917a8554606eba138a7bb3fdeaa9c
Parents: 98c7581
Author: kunal642 <kunalkapoor642@gmail.com>
Authored: Tue Jul 10 14:38:05 2018 +0530
Committer: Venkata Ramana G <ramana.gollamudi@huawei.com>
Committed: Fri Jul 13 17:11:27 2018 +0530

----------------------------------------------------------------------
 .../carbondata/core/locks/CarbonLockFactory.java       |  4 ----
 .../apache/carbondata/core/locks/CarbonLockUtil.java   | 13 +++++++++++--
 .../carbondata/hadoop/api/CarbonFileInputFormat.java   |  4 +---
 .../main/scala/org/apache/spark/sql/CarbonEnv.scala    |  2 +-
 .../command/schema/CarbonGetTableDetailCommand.scala   | 12 ++++--------
 .../command/table/CarbonCreateTableCommand.scala       |  3 ++-
 .../command/table/CarbonDropTableCommand.scala         |  8 ++++----
 .../apache/spark/sql/hive/CarbonFileMetastore.scala    |  3 ++-
 .../org/apache/spark/sql/hive/CarbonSessionState.scala |  3 ++-
 9 files changed, 27 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockFactory.java b/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockFactory.java
index 769e752..91677a6 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockFactory.java
@@ -60,10 +60,6 @@ public class CarbonLockFactory {
     if (lockPath.isEmpty()) {
       absoluteLockPath = absoluteTableIdentifier.getTablePath();
     } else {
-      if (absoluteTableIdentifier
-          .getCarbonTableIdentifier().getTableId().isEmpty()) {
-        throw new RuntimeException("Table id is empty");
-      }
       absoluteLockPath =
           getLockpath(absoluteTableIdentifier.getCarbonTableIdentifier().getTableId());
     }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockUtil.java b/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockUtil.java
index 4d67faf..ca6cddb 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/CarbonLockUtil.java
@@ -19,6 +19,7 @@ package org.apache.carbondata.core.locks;
 
 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.filesystem.CarbonFileFilter;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
@@ -27,6 +28,8 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
 import org.apache.carbondata.core.util.CarbonProperties;
 import org.apache.carbondata.core.util.path.CarbonTablePath;
 
+import org.apache.commons.lang.StringUtils;
+
 /**
  * This class contains all carbon lock utilities
  */
@@ -121,8 +124,14 @@ public class CarbonLockUtil {
     final long segmentLockFilesPreservTime =
         CarbonProperties.getInstance().getSegmentLockFilesPreserveHours();
     AbsoluteTableIdentifier absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier();
-    String lockFilesDir = CarbonTablePath
-        .getLockFilesDirPath(absoluteTableIdentifier.getTablePath());
+    String lockFilesDir = CarbonProperties.getInstance()
+        .getProperty(CarbonCommonConstants.LOCK_PATH, "");
+    if (StringUtils.isEmpty(lockFilesDir)) {
+      lockFilesDir = CarbonTablePath.getLockFilesDirPath(absoluteTableIdentifier.getTablePath());
+    } else {
+      lockFilesDir = CarbonTablePath.getLockFilesDirPath(
+          CarbonLockFactory.getLockpath(carbonTable.getTableInfo().getFactTable().getTableId()));
+    }
     CarbonFile[] files = FileFactory.getCarbonFile(lockFilesDir)
         .listFiles(new CarbonFileFilter() {
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
index 8755176..0f02e12 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
@@ -100,13 +100,11 @@ public class CarbonFileInputFormat<T> extends CarbonInputFormat<T>
implements Se
    */
   @Override
   public List<InputSplit> getSplits(JobContext job) throws IOException {
-
-    AbsoluteTableIdentifier identifier = getAbsoluteTableIdentifier(job.getConfiguration());
     CarbonTable carbonTable = getOrCreateCarbonTable(job.getConfiguration());
-
     if (null == carbonTable) {
       throw new IOException("Missing/Corrupt schema file for table.");
     }
+    AbsoluteTableIdentifier identifier = carbonTable.getAbsoluteTableIdentifier();
 
     if (getValidateSegmentsToAccess(job.getConfiguration())) {
       // get all valid segments and set them into the configuration

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
index 5650187..70c4f12 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
@@ -226,7 +226,7 @@ object CarbonEnv {
       DataMapStoreManager.getInstance().
         clearDataMaps(AbsoluteTableIdentifier.from(tablePath,
           identifier.database.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase),
-          identifier.table))
+          identifier.table, table.get.getTableInfo.getFactTable.getTableId))
       isRefreshed = true
     }
     isRefreshed

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonGetTableDetailCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonGetTableDetailCommand.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonGetTableDetailCommand.scala
index 2c702b6..8d6a4cc 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonGetTableDetailCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonGetTableDetailCommand.scala
@@ -35,20 +35,16 @@ case class CarbonGetTableDetailCommand(
   extends DataCommand {
 
   override def processData(sparkSession: SparkSession): Seq[Row] = {
-    val storePath = CarbonProperties.getStorePath
     if (tableNames.isDefined) {
       tableNames.get.map { tablename =>
-        val absoluteTableIdentifier =
-          AbsoluteTableIdentifier.from(storePath, databaseName.toLowerCase, tablename.toLowerCase)
-        val carbonTableIdentifier = absoluteTableIdentifier.getCarbonTableIdentifier
-        val carbonTable = CarbonEnv.getCarbonTable(Option(carbonTableIdentifier.getDatabaseName),
-          carbonTableIdentifier.getTableName)(sparkSession)
+        val carbonTable = CarbonEnv.getCarbonTable(Option(databaseName),
+          tablename)(sparkSession)
 
         Row(
           tablename,
           carbonTable.size,
-          SegmentStatusManager.getTableStatusLastModifiedTime(absoluteTableIdentifier)
-        )
+          SegmentStatusManager
+            .getTableStatusLastModifiedTime(carbonTable.getAbsoluteTableIdentifier))
       }
     } else {
       Seq.empty[Row]

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
index 543ba39..c403d52 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
@@ -81,7 +81,8 @@ case class CarbonCreateTableCommand(
         throw new UnsupportedOperationException("streaming is not supported with s3 store")
       }
       tableInfo.setTablePath(tablePath)
-      val tableIdentifier = AbsoluteTableIdentifier.from(tablePath, dbName, tableName)
+      val tableIdentifier = AbsoluteTableIdentifier
+        .from(tablePath, dbName, tableName, tableInfo.getFactTable.getTableId)
 
       // Add validation for sort scope when create table
       val sortScope = tableInfo.getFactTable.getTableProperties.asScala

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
index 5d74c2c..e4b298f 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
@@ -53,9 +53,7 @@ case class CarbonDropTableCommand(
   override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
     val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
 
-
-    val identifier = CarbonEnv.getIdentifier(databaseNameOp, tableName)(sparkSession)
-    val dbName = identifier.getCarbonTableIdentifier.getDatabaseName
+    val dbName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
     val carbonLocks: scala.collection.mutable.ListBuffer[ICarbonLock] = ListBuffer()
     try {
       carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
@@ -64,8 +62,10 @@ case class CarbonDropTableCommand(
       } else {
         List.empty
       }
+      val identifier = carbonTable.getAbsoluteTableIdentifier
       locksToBeAcquired foreach {
-        lock => carbonLocks += CarbonLockUtil.getLockObject(identifier, lock)
+        lock => carbonLocks +=
+                CarbonLockUtil.getLockObject(identifier, lock)
       }
 
       if (SegmentStatusManager.isLoadInProgressInTable(carbonTable)) {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
index 1670d8a..dddc72c 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
@@ -355,7 +355,8 @@ class CarbonFileMetastore extends CarbonMetaStore {
     val tableName = tableInfo.getFactTable.getTableName
     val thriftTableInfo = schemaConverter.fromWrapperToExternalTableInfo(
       tableInfo, dbName, tableName)
-    val identifier = AbsoluteTableIdentifier.from(tablePath, dbName, tableName)
+    val identifier = AbsoluteTableIdentifier
+      .from(tablePath, dbName, tableName, thriftTableInfo.getFact_table.getTable_id)
     createSchemaThriftFile(identifier, thriftTableInfo)
     LOGGER.info(s"Table $tableName for Database $dbName created successfully.")
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/637a9746/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
index b361e36..2c98ec2 100644
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
+++ b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
@@ -176,7 +176,8 @@ class CarbonHiveSessionCatalog(
       refreshTable(identifier)
       DataMapStoreManager.getInstance().
         clearDataMaps(AbsoluteTableIdentifier.from(storePath,
-          identifier.database.getOrElse("default"), identifier.table))
+          identifier.database.getOrElse("default"),
+          identifier.table))
       isRefreshed = true
       logInfo(s"Schema changes have been detected for table: $identifier")
     }


Mime
View raw message