[CARBONDATA-1719][Pre-Aggregate][Bug] Fixed bug to handle data incons…

…istency on concurrent data load and pre-aggregate table creation

Problem: On concurrent data load and pre-aggregate table creation, datamap was not getting populated with the load data even after data load completes.

This closes #1697

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

@@ -20,9 +20,10 @@ package org.apache.spark.sql.execution.command.preaaggregate
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 
+import org.apache.spark.sql.CarbonEnv
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.execution.command.management.CarbonAlterTableCompactionCommand
 import org.apache.spark.sql.execution.command.AlterTableModel
+import org.apache.spark.sql.execution.command.management.CarbonAlterTableCompactionCommand
 
 import org.apache.carbondata.core.metadata.schema.table.AggregationDataMapSchema
 import org.apache.carbondata.core.util.CarbonUtil
@@ -38,7 +39,8 @@ object LoadPostAggregateListener extends OperationEventListener {
     val loadEvent = event.asInstanceOf[LoadTablePreStatusUpdateEvent]
     val sparkSession = loadEvent.sparkSession
     val carbonLoadModel = loadEvent.carbonLoadModel
-    val table = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+    val table = CarbonEnv.getCarbonTable(Option(carbonLoadModel.getDatabaseName),
+      carbonLoadModel.getTableName)(sparkSession)
     if (CarbonUtil.hasAggregationDataMap(table)) {
       // getting all the aggergate datamap schema
       val aggregationDataMapList = table.getTableInfo.getDataMapSchemaList.asScala