From 94edb0d78d816c8fa108e761c3313c588dc794f1 Mon Sep 17 00:00:00 2001
From: Sssan520 <liangap2008@aliyun.com>
Date: Mon, 2 Jul 2018 19:12:24 +0800
Subject: [PATCH] [CARBONDATA-2819] Fixed cannot drop preagg datamap on table
 if the table has other type datamaps

---
 .../preaggregate/TestPreAggregateDrop.scala   | 32 +++++++++++++++++++
 .../datamap/CarbonDropDataMapCommand.scala    | 15 +++++----
 2 files changed, 41 insertions(+), 6 deletions(-)

diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
index f73a5871b9f..6495475c378 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
@@ -129,6 +129,38 @@ class TestPreAggregateDrop extends QueryTest with BeforeAndAfterAll {
     assert(e.getMessage.contains("Table or view 'maintable' not found in"))
   }
 
+  test("drop preaggregate datamap which its main table has other type datamaps") {
+    sql("drop table if exists maintable")
+    sql("create table maintable (a string, b string, c string) stored by 'carbondata'")
+    sql(
+      "create datamap bloom1 on table maintable using 'bloomfilter'" +
+      " dmproperties('index_columns'='a')")
+    sql(
+      "create datamap preagg1 on table maintable using 'preaggregate' as select" +
+      " a,sum(b) from maintable group by a")
+    sql("drop datamap if exists bloom1 on table maintable")
+    sql("drop datamap if exists preagg1 on table maintable")
+    checkExistence(sql("SHOW DATAMAP ON TABLE maintable"), false, "maintable_preagg1")
+    sql(
+      "create datamap bloom1 on table maintable using 'bloomfilter'" +
+      " dmproperties('index_columns'='a')")
+    sql(
+      "create datamap preagg1 on table maintable using 'preaggregate' as select" +
+      " a,sum(b) from maintable group by a")
+    sql("drop datamap if exists preagg1 on table maintable")
+    sql("drop datamap if exists bloom1 on table maintable")
+    checkExistence(sql("SHOW DATAMAP ON TABLE maintable"), false, "maintable_preagg1")
+    sql(
+      "create datamap bloom1 on table maintable using 'bloomfilter'" +
+      " dmproperties('index_columns'='a')")
+    sql(
+      "create datamap preagg1 on table maintable using 'preaggregate' as select" +
+      " a,sum(b) from maintable group by a")
+    sql("drop table if exists maintable")
+    checkExistence(sql("show tables"), false, "maintable_preagg1", "maintable")
+    checkExistence(sql("show datamap"), false, "bloom1")
+  }
+
   override def afterAll() {
     sql("drop table if exists maintable")
     sql("drop table if exists maintable1")
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
index 8bc269b7821..e0aea6b1702 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
@@ -116,8 +116,12 @@ case class CarbonDropDataMapCommand(
         // drop index datamap on the main table
         if (mainTable != null &&
             DataMapStoreManager.getInstance().getAllDataMap(mainTable).size() > 0) {
-          dropDataMapFromSystemFolder(sparkSession)
-          return Seq.empty
+          val dmSchemaOp = DataMapStoreManager.getInstance().getAllDataMap(mainTable).asScala.
+            find(_.getDataMapSchema.getDataMapName.equalsIgnoreCase(dataMapName))
+          if (dmSchemaOp.isDefined) {
+            dropDataMapFromSystemFolder(sparkSession)
+            return Seq.empty
+          }
         }
 
         // If datamap to be dropped in parent table then drop the datamap from metastore and remove
@@ -160,9 +164,8 @@ case class CarbonDropDataMapCommand(
           } else if (!ifExistsSet) {
             throw new NoSuchDataMapException(dataMapName, tableName)
           }
-        } else if (mainTable != null &&
-                   mainTable.getTableInfo.getDataMapSchemaList.size() == 0) {
-          dropDataMapFromSystemFolder(sparkSession)
+        } else if (!ifExistsSet) {
+          throw new NoSuchDataMapException(dataMapName)
         }
       } catch {
         case e: NoSuchDataMapException =>
@@ -191,7 +194,7 @@ case class CarbonDropDataMapCommand(
       }
     }
 
-      Seq.empty
+    Seq.empty
   }
 
   private def dropDataMapFromSystemFolder(sparkSession: SparkSession): Unit = {