[CARBONDATA-3284] [CARBONDATA-3285] Workaround for Create-PreAgg Data…

…map Fail & Sort-Columns Fix

If for some reason**[1]**, creating PreAgg datamap failed and its dropping also failed.
Then dropping datamap also cannot be done, as the datamap was not registered to the parent table schema file, but got registered in spark-hive, so it shows it as a table, but won't let us drop it as carbon throws error if we try to drop it as a table.

Workaround:
After this change, we can at lease drop that as a hive folder by command

**[1]** - Reason could be something like setting HDFS Quota on database folder, so that parent table schema file cound not be modified.

*

This closes #3113

core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java

@@ -3348,4 +3348,20 @@ public static int[] getInvertedReverseIndex(int[] invertedIndex) {
   public static String generateUUID() {
     return UUID.randomUUID().toString();
   }
+
+  /**
+   * Below method will be used to get the datamap schema name from datamap table name
+   * it will split name based on character '_' and get the last name
+   * This is only for pre aggregate and timeseries tables
+   *
+   * @param tableName
+   * @return datamapschema name
+   */
+  public static String getDatamapNameFromTableName(String tableName) {
+    int i = tableName.lastIndexOf('_');
+    if (i != -1) {
+      return tableName.substring(i + 1, tableName.length());
+    }
+    return null;
+  }
 }

docs/faq.md

@@ -43,6 +43,7 @@
 - [Failed to insert data on the cluster](#failed-to-insert-data-on-the-cluster)
 - [Failed to execute Concurrent Operations(Load,Insert,Update) on table by multiple workers](#failed-to-execute-concurrent-operations-on-table-by-multiple-workers)
 - [Failed to create a table with a single numeric column](#failed-to-create-a-table-with-a-single-numeric-column)
+- [Failed to create datamap and drop datamap is also not working](#failed-to-create-datamap-and-drop-datamap-is-also-not-working)
 
 ## 
 
@@ -474,4 +475,21 @@ Note : Refrain from using "mvn clean package" without specifying the profile.
 
   A single column that can be considered as dimension is mandatory for table creation.
 
+## Failed to create datamap and drop datamap is also not working
+
+  **Symptom**
+
+  Execution fails with the following exception :
+
+  ```
+  HDFS Quota Exceeded
+  ```
+
+  **Possible Cause**
+
+  HDFS Quota is set, and it is not letting carbondata write or modify any files.
+
+  **Procedure**
 
+  Drop that particular datamap using Drop Table command using table name as
+  parentTableName_datamapName so as to clear the stale folders.

integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/QueriesBVATestCase.scala

@@ -10697,8 +10697,8 @@ class QueriesBVATestCase extends QueryTest with BeforeAndAfterAll {
   //PushUP_FILTER_test_boundary_TC194
   test("PushUP_FILTER_test_boundary_TC194", Include) {
 
-    checkAnswer(s"""select min(c2_Bigint),max(c2_Bigint),sum(c2_Bigint),avg(c2_Bigint) , count(c2_Bigint), variance(c2_Bigint) from Test_Boundary where sin(c1_int)=0.18796200317975467 or sin(c1_int)=-0.18796200317975467""",
-      s"""select min(c2_Bigint),max(c2_Bigint),sum(c2_Bigint),avg(c2_Bigint) , count(c2_Bigint), variance(c2_Bigint) from Test_Boundary_hive where sin(c1_int)=0.18796200317975467 or sin(c1_int)=-0.18796200317975467""", "QueriesBVATestCase_PushUP_FILTER_test_boundary_TC194")
+    checkAnswer(s"""select min(c2_Bigint),max(c2_Bigint),sum(c2_Bigint),avg(c2_Bigint) , count(c2_Bigint), variance(c2_Bigint) from (select c2_Bigint from Test_Boundary where sin(c1_int)=0.18796200317975467 or sin(c1_int)=-0.18796200317975467 order by c2_Bigint""",
+      s"""select min(c2_Bigint),max(c2_Bigint),sum(c2_Bigint),avg(c2_Bigint) , count(c2_Bigint), variance(c2_Bigint) from (select c2_Bigint from Test_Boundary_hive where sin(c1_int)=0.18796200317975467 or sin(c1_int)=-0.18796200317975467 order by c2_Bigint""", "QueriesBVATestCase_PushUP_FILTER_test_boundary_TC194")
 
   }
 

integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/suite/SDVSuites.scala

@@ -83,6 +83,7 @@ class SDVSuites extends Suites with BeforeAndAfterAll {
 class SDVSuites1 extends Suites with BeforeAndAfterAll {
 
   val suites =     new BadRecordTestCase ::
+                   new ComplexDataTypeTestCase ::
                    new BatchSortLoad1TestCase ::
                    new BatchSortQueryTestCase ::
                    new DataLoadingTestCase ::
@@ -156,7 +157,6 @@ class SDVSuites3 extends Suites with BeforeAndAfterAll {
                     new TestPartitionWithGlobalSort ::
                     new SDKwriterTestCase ::
                     new SetParameterTestCase ::
-                    new ComplexDataTypeTestCase ::
                     new PartitionWithPreAggregateTestCase ::
                     new CreateTableWithLocalDictionaryTestCase ::
                     new LoadTableWithLocalDictionaryTestCase :: Nil

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

@@ -206,7 +206,7 @@ case class CarbonLoadDataCommand(
     * 4. Session property CARBON_OPTIONS_SORT_SCOPE
     * 5. Default Sort Scope LOAD_SORT_SCOPE
     */
-    if (StringUtils.isBlank(tableProperties.get(CarbonCommonConstants.SORT_COLUMNS))) {
+    if (table.getNumberOfSortColumns == 0) {
       // If tableProperties.SORT_COLUMNS is null
       optionsFinal.put(CarbonCommonConstants.SORT_SCOPE,
         SortScopeOptions.SortScope.NO_SORT.name)

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

@@ -127,8 +127,6 @@ private[sql] case class CarbonAlterTableRenameCommand(
       schemaEvolutionEntry.setTime_stamp(timeStamp)
       val newCarbonTableIdentifier = new CarbonTableIdentifier(oldDatabaseName,
         newTableName, carbonTable.getCarbonTableIdentifier.getTableId)
-      val oldIdentifier = TableIdentifier(oldTableName, Some(oldDatabaseName))
-      val newIdentifier = TableIdentifier(newTableName, Some(oldDatabaseName))
       metastore.removeTableFromMetadata(oldDatabaseName, oldTableName)
       var partitions: Seq[CatalogTablePartition] = Seq.empty
       if (carbonTable.isHivePartitionTable) {

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

@@ -77,7 +77,37 @@ case class CarbonDropTableCommand(
       }
       val relationIdentifiers = carbonTable.getTableInfo.getParentRelationIdentifiers
       if (relationIdentifiers != null && !relationIdentifiers.isEmpty) {
-        if (!dropChildTable) {
+        var ignoreParentTableCheck = false
+        if (carbonTable.getTableInfo.getParentRelationIdentifiers.size() == 1) {
+          /**
+           * below handling in case when pre aggregation creation failed in scenario
+           * while creating a pre aggregate data map it created pre aggregate table and registered
+           * in hive, but failed to register in main table because of some exception.
+           * in this case if it will not allow user to drop datamap and data map table
+           * for this if user run drop table command for pre aggregate it should allow user to drop
+           * the same
+           */
+          val parentDbName =
+            carbonTable.getTableInfo.getParentRelationIdentifiers.get(0).getDatabaseName
+          val parentTableName =
+            carbonTable.getTableInfo.getParentRelationIdentifiers.get(0).getTableName
+          val parentCarbonTable = try {
+            Some(CarbonEnv.getCarbonTable(Some(parentDbName), parentTableName)(sparkSession))
+          } catch {
+            case _: Exception => None
+          }
+          if (parentCarbonTable.isDefined) {
+            val dataMapSchemaName = CarbonUtil.getDatamapNameFromTableName(carbonTable.getTableName)
+            if (null != dataMapSchemaName) {
+              val dataMapSchema = parentCarbonTable.get.getDataMapSchema(dataMapSchemaName)
+              if (null == dataMapSchema) {
+                LOGGER.info(s"Force dropping datamap ${carbonTable.getTableName}")
+                ignoreParentTableCheck = true
+              }
+            }
+          }
+        }
+        if (!ignoreParentTableCheck && !dropChildTable) {
           if (!ifExistsSet) {
             throwMetadataException(dbName, tableName,
               "Child table which is associated with datamap cannot be dropped, " +