carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From manishgupt...@apache.org
Subject [09/50] [abbrv] carbondata git commit: [CARBONDATA-2416] Support DEFERRED REBUILD when creating DataMap
Date Thu, 17 May 2018 14:30:22 GMT
http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
index 67effda..cccfb3f 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapStatus.scala
@@ -17,6 +17,7 @@
 
 package org.apache.carbondata.spark.testsuite.datamap
 
+import java.io.File
 import java.util
 
 import scala.collection.JavaConverters._
@@ -26,14 +27,12 @@ import org.scalatest.BeforeAndAfterAll
 
 import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException
 import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datamap.dev.{DataMapRefresher, DataMapWriter}
 import org.apache.carbondata.core.datamap.dev.cgdatamap.{CoarseGrainDataMap, CoarseGrainDataMapFactory}
+import org.apache.carbondata.core.datamap.dev.{DataMapBuilder, DataMapWriter}
 import org.apache.carbondata.core.datamap.status.{DataMapStatus, DataMapStatusManager}
 import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, Segment}
 import org.apache.carbondata.core.datastore.page.ColumnPage
 import org.apache.carbondata.core.features.TableOperation
-import org.apache.carbondata.core.datastore.row.CarbonRow
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
 import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
 import org.apache.carbondata.core.scan.filter.intf.ExpressionType
 import org.apache.carbondata.core.util.CarbonProperties
@@ -44,10 +43,11 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
   val testData = s"$resourcesPath/sample.csv"
 
   override def beforeAll: Unit = {
+    new File(CarbonProperties.getInstance().getSystemFolderLocation).delete()
     drop
   }
 
-  test("datamap status disable for new datamap") {
+  test("datamap status enable for new datamap") {
     sql("DROP TABLE IF EXISTS datamapstatustest")
     sql(
       """
@@ -64,11 +64,33 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
 
     assert(details.length == 1)
 
+    assert(details.exists(p => p.getDataMapName.equals("statusdatamap") && p.getStatus == DataMapStatus.ENABLED))
+    sql("DROP TABLE IF EXISTS datamapstatustest")
+  }
+
+  test("datamap status disable for new datamap with deferred rebuild") {
+    sql("DROP TABLE IF EXISTS datamapstatustest")
+    sql(
+      """
+        | CREATE TABLE datamapstatustest(id int, name string, city string, age int)
+        | STORED BY 'org.apache.carbondata.format'
+      """.stripMargin)
+    sql(
+      s"""create datamap statusdatamap on table datamapstatustest
+         |using '${classOf[TestDataMapFactory].getName}'
+         |with deferred rebuild
+         |dmproperties('index_columns'='name')
+         | """.stripMargin)
+
+    val details = DataMapStatusManager.readDataMapStatusDetails()
+
+    assert(details.length == 1)
+
     assert(details.exists(p => p.getDataMapName.equals("statusdatamap") && p.getStatus == DataMapStatus.DISABLED))
     sql("DROP TABLE IF EXISTS datamapstatustest")
   }
 
-  test("datamap status disable after new load") {
+  test("datamap status disable after new load  with deferred rebuild") {
     sql("DROP TABLE IF EXISTS datamapstatustest1")
     sql(
       """
@@ -78,6 +100,7 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
     sql(
       s"""create datamap statusdatamap1 on table datamapstatustest1
          |using '${classOf[TestDataMapFactory].getName}'
+         |with deferred rebuild
          |dmproperties('index_columns'='name')
          | """.stripMargin)
 
@@ -94,8 +117,7 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
     sql("DROP TABLE IF EXISTS datamapstatustest1")
   }
 
-  // enable it in PR2255
-  ignore("datamap status with refresh datamap") {
+  test("datamap status with REBUILD DATAMAP") {
     sql("DROP TABLE IF EXISTS datamapstatustest2")
     sql(
       """
@@ -105,6 +127,7 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
     sql(
       s"""create datamap statusdatamap2 on table datamapstatustest2
          |using '${classOf[TestDataMapFactory].getName}'
+         |with deferred rebuild
          |dmproperties('index_columns'='name')
          | """.stripMargin)
 
@@ -119,7 +142,7 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
     assert(details.length == 1)
     assert(details.exists(p => p.getDataMapName.equals("statusdatamap2") && p.getStatus == DataMapStatus.DISABLED))
 
-    sql(s"refresh datamap statusdatamap2 on table datamapstatustest2")
+    sql(s"REBUILD DATAMAP statusdatamap2 on table datamapstatustest2")
 
     details = DataMapStatusManager.readDataMapStatusDetails()
     assert(details.length == 1)
@@ -128,8 +151,7 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
     sql("DROP TABLE IF EXISTS datamapstatustest2")
   }
 
-  // enable it in PR2255
-  ignore("datamap create without on table test") {
+  test("datamap create without on table test") {
     sql("DROP TABLE IF EXISTS datamapstatustest3")
     sql(
       """
@@ -144,10 +166,20 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
            | """.stripMargin)
 
     }
+    sql("DROP TABLE IF EXISTS datamapstatustest3")
+  }
 
+  test("rebuild datamap status") {
+    sql("DROP TABLE IF EXISTS datamapstatustest3")
+    sql(
+      """
+        | CREATE TABLE datamapstatustest3(id int, name string, city string, age int)
+        | STORED BY 'org.apache.carbondata.format'
+      """.stripMargin)
     sql(
       s"""create datamap statusdatamap3 on table datamapstatustest3
          |using '${classOf[TestDataMapFactory].getName}'
+         |with deferred rebuild
          |dmproperties('index_columns'='name')
          | """.stripMargin)
 
@@ -162,7 +194,7 @@ class TestDataMapStatus extends QueryTest with BeforeAndAfterAll {
     assert(details.length == 1)
     assert(details.exists(p => p.getDataMapName.equals("statusdatamap3") && p.getStatus == DataMapStatus.DISABLED))
 
-    sql(s"refresh datamap statusdatamap3")
+    sql(s"REBUILD DATAMAP statusdatamap3")
 
     details = DataMapStatusManager.readDataMapStatusDetails()
     assert(details.length == 1)
@@ -245,8 +277,19 @@ class TestDataMapFactory(
     false
   }
 
-  override def createRefresher(segment: Segment,
-      shardName: String): DataMapRefresher = {
-    ???
+  override def createBuilder(segment: Segment,
+      shardName: String): DataMapBuilder = {
+    return new DataMapBuilder {
+      override def initialize(): Unit = { }
+
+      override def addRow(blockletId: Int,
+          pageId: Int,
+          rowId: Int,
+          values: Array[AnyRef]): Unit = { }
+
+      override def finish(): Unit = { }
+
+      override def close(): Unit = { }
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SearchModeTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SearchModeTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SearchModeTestCase.scala
index c193fcf..2f3488e 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SearchModeTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SearchModeTestCase.scala
@@ -28,8 +28,6 @@ import org.apache.carbondata.spark.util.DataGenerator
  * Test Suite for search mode
  */
 
-// TODO: Need to Fix
-@Ignore
 class SearchModeTestCase extends QueryTest with BeforeAndAfterAll {
 
   val numRows = 500 * 1000

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
index 9c5297d..3cabc7b 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertAndOtherCommandConcurrent.scala
@@ -24,11 +24,12 @@ import java.util.concurrent.{Callable, ExecutorService, Executors, Future}
 import scala.collection.JavaConverters._
 
 import org.apache.spark.sql.test.util.QueryTest
+import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.{DataFrame, SaveMode}
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datamap.dev.{DataMapRefresher, DataMapWriter}
+import org.apache.carbondata.core.datamap.dev.{DataMapBuilder, DataMapWriter}
 import org.apache.carbondata.core.datamap.dev.cgdatamap.{CoarseGrainDataMap, CoarseGrainDataMapFactory}
 import org.apache.carbondata.core.datamap.{DataMapDistributable, DataMapMeta, Segment}
 import org.apache.carbondata.core.datastore.page.ColumnPage
@@ -43,18 +44,30 @@ import org.apache.carbondata.events.Event
 // This testsuite test insert and insert overwrite with other commands concurrently
 class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterAll with BeforeAndAfterEach {
   private val executorService: ExecutorService = Executors.newFixedThreadPool(10)
-  var df: DataFrame = _
+  var testData: DataFrame = _
 
   override def beforeAll {
     dropTable()
     buildTestData()
 
+    createTable("orders", testData.schema)
+    createTable("orders_overwrite", testData.schema)
     sql(
       s"""
          | create datamap test on table orders
          | using '${classOf[WaitingDataMapFactory].getName}'
          | dmproperties('index_columns'='o_name')
        """.stripMargin)
+
+    testData.write
+      .format("carbondata")
+      .option("tableName", "temp_table")
+      .option("tempCSV", "false")
+      .mode(SaveMode.Overwrite)
+      .save()
+
+    sql(s"insert into orders select * from temp_table")
+    sql(s"insert into orders_overwrite select * from temp_table")
   }
 
   private def buildTestData(): Unit = {
@@ -66,23 +79,17 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
     import sqlContext.implicits._
 
     val sdf = new SimpleDateFormat("yyyy-MM-dd")
-    df = sqlContext.sparkSession.sparkContext.parallelize(1 to 150000)
+    testData = sqlContext.sparkSession.sparkContext.parallelize(1 to 150000)
       .map(value => (value, new java.sql.Date(sdf.parse("2015-07-" + (value % 10 + 10)).getTime),
         "china", "aaa" + value, "phone" + 555 * value, "ASD" + (60000 + value), 14999 + value,
         "ordersTable" + value))
       .toDF("o_id", "o_date", "o_country", "o_name",
         "o_phonetype", "o_serialname", "o_salary", "o_comment")
-    createTable("orders")
-    createTable("orders_overwrite")
   }
 
-  private def createTable(tableName: String): Unit = {
-    df.write
-      .format("carbondata")
-      .option("tableName", tableName)
-      .option("tempCSV", "false")
-      .mode(SaveMode.Overwrite)
-      .save()
+  private def createTable(tableName: String, schema: StructType): Unit = {
+    val schemaString = schema.fields.map(x => x.name + " " + x.dataType.typeName).mkString(", ")
+    sql(s"CREATE TABLE $tableName ($schemaString) stored as carbondata")
   }
 
   override def afterAll {
@@ -91,7 +98,7 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
   }
 
   override def beforeEach(): Unit = {
-    Global.overwriteRunning = false
+    Global.loading = false
   }
 
   private def dropTable() = {
@@ -101,12 +108,12 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
 
   // run the input SQL and block until it is running
   private def runSqlAsync(sql: String): Future[String] = {
-    assert(!Global.overwriteRunning)
+    assert(!Global.loading)
     var count = 0
     val future = executorService.submit(
       new QueryTask(sql)
     )
-    while (!Global.overwriteRunning && count < 1000) {
+    while (!Global.loading && count < 1000) {
       Thread.sleep(10)
       // to avoid dead loop in case WaitingDataMapFactory is not invoked
       count += 1
@@ -202,9 +209,7 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
     sql("drop table if exists t1")
 
     // number of segment is 1 after createTable
-    createTable("t1")
-    // number of segment is 2 after insert
-    sql("insert into table t1 select * from orders_overwrite")
+    createTable("t1", testData.schema)
 
     sql(
       s"""
@@ -212,6 +217,12 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
          | using '${classOf[WaitingDataMapFactory].getName}'
          | dmproperties('index_columns'='o_name')
        """.stripMargin)
+
+    sql("insert into table t1 select * from orders_overwrite")
+    Thread.sleep(1100)
+    sql("insert into table t1 select * from orders_overwrite")
+    Thread.sleep(1100)
+
     val future = runSqlAsync("insert into table t1 select * from orders_overwrite")
     sql("alter table t1 compact 'MAJOR'")
     assert(future.get.contains("PASS"))
@@ -279,15 +290,13 @@ class TestInsertAndOtherCommandConcurrent extends QueryTest with BeforeAndAfterA
 }
 
 object Global {
-  var overwriteRunning = false
+  var loading = false
 }
 
 class WaitingDataMapFactory(
     carbonTable: CarbonTable,
     dataMapSchema: DataMapSchema) extends CoarseGrainDataMapFactory(carbonTable, dataMapSchema) {
 
-  private var identifier: AbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
-
   override def fireEvent(event: Event): Unit = ???
 
   override def clear(segmentId: Segment): Unit = {}
@@ -311,14 +320,14 @@ class WaitingDataMapFactory(
 
       override def onBlockStart(blockId: String): Unit = {
         // trigger the second SQL to execute
-        Global.overwriteRunning = true
+        Global.loading = true
 
         // wait for 1 second to let second SQL to finish
         Thread.sleep(1000)
       }
 
       override def finish(): Unit = {
-
+        Global.loading = false
       }
     }
   }
@@ -341,8 +350,8 @@ class WaitingDataMapFactory(
     false
   }
 
-  override def createRefresher(segment: Segment,
-      shardName: String): DataMapRefresher = {
+  override def createBuilder(segment: Segment,
+      shardName: String): DataMapBuilder = {
     ???
   }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
index fea3482..890f8fc 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
@@ -181,7 +181,8 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser {
   protected val ON = carbonKeyWord("ON")
   protected val DMPROPERTIES = carbonKeyWord("DMPROPERTIES")
   protected val SELECT = carbonKeyWord("SELECT")
-  protected val REFRESH = carbonKeyWord("REFRESH")
+  protected val REBUILD = carbonKeyWord("REBUILD")
+  protected val DEFERRED = carbonKeyWord("DEFERRED")
 
   protected val doubleQuotedString = "\"([^\"]+)\"".r
   protected val singleQuotedString = "'([^']+)'".r

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapProperty.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapProperty.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapProperty.java
deleted file mode 100644
index 043acb1..0000000
--- a/integration/spark2/src/main/java/org/apache/carbondata/datamap/DataMapProperty.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.carbondata.datamap;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-
-/**
- * Property that can be specified when creating DataMap
- */
-@InterfaceAudience.Internal
-public class DataMapProperty {
-
-  /**
-   * Used to specify the store location of the datamap
-   */
-  public static final String PARTITIONING = "partitioning";
-  public static final String PATH = "path";
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
index a5124a0..0642e01 100644
--- a/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
+++ b/integration/spark2/src/main/java/org/apache/carbondata/datamap/IndexDataMapProvider.java
@@ -105,7 +105,7 @@ public class IndexDataMapProvider extends DataMapProvider {
 
   @Override
   public void rebuild() {
-    IndexDataMapRefreshRDD.rebuildDataMap(sparkSession, getMainTable(), getDataMapSchema());
+    IndexDataMapRebuildRDD.rebuildDataMap(sparkSession, getMainTable(), getDataMapSchema());
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java b/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
index 746a361..37d49e5 100644
--- a/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
+++ b/integration/spark2/src/main/java/org/apache/carbondata/datamap/PreAggregateDataMapProvider.java
@@ -17,11 +17,15 @@
 
 package org.apache.carbondata.datamap;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.apache.carbondata.common.annotations.InterfaceAudience;
 import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
 import org.apache.carbondata.core.datamap.DataMapCatalog;
 import org.apache.carbondata.core.datamap.DataMapProvider;
 import org.apache.carbondata.core.datamap.dev.DataMapFactory;
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProperty;
 import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
 import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
 
@@ -55,9 +59,11 @@ public class PreAggregateDataMapProvider extends DataMapProvider {
   private void validateDmProperty(DataMapSchema dataMapSchema)
       throws MalformedDataMapCommandException {
     if (!dataMapSchema.getProperties().isEmpty()) {
-      if (dataMapSchema.getProperties().size() > 2 || (
-              !dataMapSchema.getProperties().containsKey(DataMapProperty.PATH) &&
-                      !dataMapSchema.getProperties().containsKey(DataMapProperty.PARTITIONING))) {
+      Map<String, String> properties = new HashMap<>(dataMapSchema.getProperties());
+      properties.remove(DataMapProperty.DEFERRED_REBUILD);
+      properties.remove(DataMapProperty.PATH);
+      properties.remove(DataMapProperty.PARTITIONING);
+      if (properties.size() > 0) {
         throw new MalformedDataMapCommandException(
                 "Only 'path' and 'partitioning' dmproperties are allowed for this datamap");
       }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala
new file mode 100644
index 0000000..5902783
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.datamap
+
+import java.io.{File, IOException}
+import java.text.SimpleDateFormat
+import java.util
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
+import org.apache.spark.{CarbonInputMetrics, Partition, TaskContext}
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.sql.SparkSession
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
+import org.apache.carbondata.core.datamap.{DataMapRegistry, DataMapStoreManager, Segment}
+import org.apache.carbondata.core.datamap.dev.DataMapBuilder
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
+import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, TableInfo}
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager
+import org.apache.carbondata.core.util.TaskMetricsMap
+import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit, CarbonProjection, CarbonRecordReader}
+import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
+import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport
+import org.apache.carbondata.spark.{RefreshResult, RefreshResultImpl}
+import org.apache.carbondata.spark.rdd.{CarbonRDDWithTableInfo, CarbonSparkPartition}
+import org.apache.carbondata.spark.util.SparkDataTypeConverterImpl
+
+/**
+ * Helper object to rebuild the index DataMap
+ */
+object IndexDataMapRebuildRDD {
+
+  /**
+   * Rebuild the datamap for all existing data in the table
+   */
+  def rebuildDataMap(
+      sparkSession: SparkSession,
+      carbonTable: CarbonTable,
+      schema: DataMapSchema
+  ): Unit = {
+    val tableIdentifier = carbonTable.getAbsoluteTableIdentifier
+    val segmentStatusManager = new SegmentStatusManager(tableIdentifier)
+    val validAndInvalidSegments = segmentStatusManager.getValidAndInvalidSegments()
+    val validSegments = validAndInvalidSegments.getValidSegments
+    val indexedCarbonColumns = carbonTable.getIndexedColumns(schema)
+
+    // loop all segments to rebuild DataMap
+    validSegments.asScala.foreach { segment =>
+      // if lucene datamap folder is exists, not require to build lucene datamap again
+      refreshOneSegment(sparkSession, carbonTable, schema.getDataMapName,
+        indexedCarbonColumns, segment.getSegmentNo);
+    }
+  }
+
+  private def refreshOneSegment(
+      sparkSession: SparkSession,
+      carbonTable: CarbonTable,
+      dataMapName: String,
+      indexColumns: java.util.List[CarbonColumn],
+      segmentId: String): Unit = {
+
+    val dataMapStorePath =
+      CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId) +
+      File.separator +
+      dataMapName
+
+    if (!FileFactory.isFileExist(dataMapStorePath)) {
+      if (FileFactory.mkdirs(dataMapStorePath, FileFactory.getFileType(dataMapStorePath))) {
+        try {
+          val status = new IndexDataMapRebuildRDD[String, Boolean](
+            sparkSession,
+            new RefreshResultImpl(),
+            carbonTable.getTableInfo,
+            dataMapName,
+            indexColumns.asScala.toArray,
+            segmentId
+          ).collect()
+
+          status.find(_._2 == false).foreach { task =>
+            throw new Exception(
+              s"Task Failed to rebuild datamap $dataMapName on segment_$segmentId")
+          }
+        } catch {
+          case ex: Throwable =>
+            // process failure
+            FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(dataMapStorePath))
+            throw new Exception(
+              s"Failed to refresh datamap $dataMapName on segment_$segmentId", ex)
+        }
+      } else {
+        throw new IOException(s"Failed to create directory $dataMapStorePath")
+      }
+    }
+  }
+
+}
+
+class OriginalReadSupport(dataTypes: Array[DataType]) extends CarbonReadSupport[Array[Object]] {
+  override def initialize(carbonColumns: Array[CarbonColumn],
+      carbonTable: CarbonTable): Unit = {
+  }
+
+  override def readRow(data: Array[Object]): Array[Object] = {
+    dataTypes.zipWithIndex.foreach { case (dataType, i) =>
+      if (dataType == DataTypes.STRING) {
+        data(i) = data(i).toString
+      }
+    }
+    data
+  }
+
+  override def close(): Unit = {
+  }
+}
+
+class IndexDataMapRebuildRDD[K, V](
+    session: SparkSession,
+    result: RefreshResult[K, V],
+    @transient tableInfo: TableInfo,
+    dataMapName: String,
+    indexColumns: Array[CarbonColumn],
+    segmentId: String
+) extends CarbonRDDWithTableInfo[(K, V)](
+  session.sparkContext, Nil, tableInfo.serialize()) {
+
+  private val dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
+  private val queryId = sparkContext.getConf.get("queryId", System.nanoTime() + "")
+  private val jobTrackerId: String = {
+    val formatter = new SimpleDateFormat("yyyyMMddHHmm")
+    formatter.format(new util.Date())
+  }
+
+  override def internalCompute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
+    val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+    val dataMapFactory =
+      DataMapManager.get().getDataMapProvider(
+        CarbonTable.buildFromTableInfo(getTableInfo), dataMapSchema, session).getDataMapFactory
+    var status = false
+    val inputMetrics = new CarbonInputMetrics
+    TaskMetricsMap.getInstance().registerThreadCallback()
+    val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value
+    inputMetrics.initBytesReadCallback(context, inputSplit)
+
+    val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
+    val attemptContext = new TaskAttemptContextImpl(new Configuration(), attemptId)
+    val format = createInputFormat(attemptContext)
+
+    val model = format.createQueryModel(inputSplit, attemptContext)
+    // one query id per table
+    model.setQueryId(queryId)
+    model.setVectorReader(false)
+    model.setForcedDetailRawQuery(false)
+    model.setRequiredRowId(true)
+
+    var reader: CarbonRecordReader[Array[Object]] = null
+    var refresher: DataMapBuilder = null
+    try {
+      reader = new CarbonRecordReader(
+        model, new OriginalReadSupport(indexColumns.map(_.getDataType)), inputMetrics)
+      reader.initialize(inputSplit, attemptContext)
+
+      // we use task name as shard name to create the folder for this datamap
+      val shardName = CarbonTablePath.getShardName(inputSplit.getAllSplits.get(0).getBlockPath)
+      refresher = dataMapFactory.createBuilder(new Segment(segmentId), shardName)
+      refresher.initialize()
+
+      var blockletId = 0
+      var firstRow = true
+      while (reader.nextKeyValue()) {
+        val rowWithPosition = reader.getCurrentValue
+        val size = rowWithPosition.length
+        val pageId = rowWithPosition(size - 2).asInstanceOf[Int]
+        val rowId = rowWithPosition(size - 1).asInstanceOf[Int]
+
+        if (!firstRow && pageId == 0 && rowId == 0) {
+          // new blocklet started, increase blockletId
+          blockletId = blockletId + 1
+        } else {
+          firstRow = false
+        }
+
+        refresher.addRow(blockletId, pageId, rowId, rowWithPosition)
+      }
+
+      refresher.finish()
+
+      status = true
+    } finally {
+      if (reader != null) {
+        try {
+          reader.close()
+        } catch {
+          case ex: Throwable =>
+            LOGGER.error(ex, "Failed to close reader")
+        }
+      }
+
+      if (refresher != null) {
+        try {
+          refresher.close()
+        } catch {
+          case ex: Throwable =>
+            LOGGER.error(ex, "Failed to close index writer")
+        }
+      }
+    }
+
+    new Iterator[(K, V)] {
+
+      var finished = false
+
+      override def hasNext: Boolean = {
+        !finished
+      }
+
+      override def next(): (K, V) = {
+        finished = true
+        result.getKey(split.index.toString, status)
+      }
+    }
+  }
+
+
+  private def createInputFormat(
+      attemptContext: TaskAttemptContextImpl) = {
+    val format = new CarbonTableInputFormat[Object]
+    val tableInfo1 = getTableInfo
+    val conf = attemptContext.getConfiguration
+    CarbonInputFormat.setTableInfo(conf, tableInfo1)
+    CarbonInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)
+    CarbonInputFormat.setTableName(conf, tableInfo1.getFactTable.getTableName)
+    CarbonInputFormat.setDataTypeConverter(conf, classOf[SparkDataTypeConverterImpl])
+
+    val identifier = tableInfo1.getOrCreateAbsoluteTableIdentifier()
+    CarbonInputFormat.setTablePath(
+      conf,
+      identifier.appendWithLocalPrefix(identifier.getTablePath))
+
+    CarbonInputFormat.setSegmentsToAccess(
+      conf,
+      Segment.toSegmentList(Array(segmentId), null))
+
+    CarbonInputFormat.setColumnProjection(
+      conf,
+      new CarbonProjection(indexColumns.map(_.getColName)))
+    format
+  }
+
+  override protected def getPartitions = {
+    if (!dataMapSchema.isIndexDataMap) {
+      throw new UnsupportedOperationException
+    }
+    val conf = new Configuration()
+    val jobConf = new JobConf(conf)
+    SparkHadoopUtil.get.addCredentials(jobConf)
+    val job = Job.getInstance(jobConf)
+    job.getConfiguration.set("query.id", queryId)
+
+    val format = new CarbonTableInputFormat[Object]
+
+    CarbonInputFormat.setSegmentsToAccess(
+      job.getConfiguration,
+      Segment.toSegmentList(Array(segmentId), null))
+
+    CarbonInputFormat.setTableInfo(
+      job.getConfiguration,
+      tableInfo)
+    CarbonInputFormat.setTablePath(
+      job.getConfiguration,
+      tableInfo.getOrCreateAbsoluteTableIdentifier().getTablePath)
+    CarbonInputFormat.setDatabaseName(
+      job.getConfiguration,
+      tableInfo.getDatabaseName)
+    CarbonInputFormat.setTableName(
+      job.getConfiguration,
+      tableInfo.getFactTable.getTableName)
+
+    format
+      .getSplits(job)
+      .asScala
+      .map(_.asInstanceOf[CarbonInputSplit])
+      .groupBy(_.taskId)
+      .map { group =>
+        new CarbonMultiBlockSplit(
+          group._2.asJava,
+          group._2.flatMap(_.getLocations).toArray)
+      }
+      .zipWithIndex
+      .map { split =>
+        new CarbonSparkPartition(id, split._2, split._1)
+      }
+      .toArray
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala
deleted file mode 100644
index c341c36..0000000
--- a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRefreshRDD.scala
+++ /dev/null
@@ -1,317 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.carbondata.datamap
-
-import java.io.{File, IOException}
-import java.text.SimpleDateFormat
-import java.util
-
-import scala.collection.JavaConverters._
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.mapred.JobConf
-import org.apache.hadoop.mapreduce.{Job, TaskAttemptID, TaskType}
-import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
-import org.apache.spark.{CarbonInputMetrics, Partition, TaskContext}
-import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.sql.SparkSession
-
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
-import org.apache.carbondata.core.datamap.dev.DataMapRefresher
-import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
-import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema, TableInfo}
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
-import org.apache.carbondata.core.statusmanager.SegmentStatusManager
-import org.apache.carbondata.core.util.TaskMetricsMap
-import org.apache.carbondata.core.util.path.CarbonTablePath
-import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit, CarbonProjection, CarbonRecordReader}
-import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
-import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport
-import org.apache.carbondata.spark.{RefreshResult, RefreshResultImpl}
-import org.apache.carbondata.spark.rdd.{CarbonRDDWithTableInfo, CarbonSparkPartition}
-import org.apache.carbondata.spark.util.SparkDataTypeConverterImpl
-
-/**
- * Helper object to rebuild the index DataMap
- */
-object IndexDataMapRefreshRDD {
-
-  /**
-   * Rebuild the datamap for all existing data in the table
-   */
-  def rebuildDataMap(
-      sparkSession: SparkSession,
-      carbonTable: CarbonTable,
-      schema: DataMapSchema
-  ): Unit = {
-    val tableIdentifier = carbonTable.getAbsoluteTableIdentifier
-    val segmentStatusManager = new SegmentStatusManager(tableIdentifier)
-    val validAndInvalidSegments = segmentStatusManager.getValidAndInvalidSegments()
-    val validSegments = validAndInvalidSegments.getValidSegments
-    val indexedCarbonColumns = carbonTable.getIndexedColumns(schema)
-
-    // loop all segments to rebuild DataMap
-    validSegments.asScala.foreach { segment =>
-      // if lucene datamap folder is exists, not require to build lucene datamap again
-      refreshOneSegment(sparkSession, carbonTable, schema.getDataMapName,
-        indexedCarbonColumns, segment.getSegmentNo);
-    }
-  }
-
-  private def refreshOneSegment(
-      sparkSession: SparkSession,
-      carbonTable: CarbonTable,
-      dataMapName: String,
-      indexColumns: java.util.List[CarbonColumn],
-      segmentId: String): Unit = {
-
-    val dataMapStorePath =
-      CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId) +
-      File.separator +
-      dataMapName
-
-    if (!FileFactory.isFileExist(dataMapStorePath)) {
-      if (FileFactory.mkdirs(dataMapStorePath, FileFactory.getFileType(dataMapStorePath))) {
-        try {
-          val status = new IndexDataMapRefreshRDD[String, Boolean](
-            sparkSession,
-            new RefreshResultImpl(),
-            carbonTable.getTableInfo,
-            dataMapName,
-            indexColumns.asScala.toArray,
-            segmentId
-          ).collect()
-
-          status.find(_._2 == false).foreach { task =>
-            throw new Exception(
-              s"Task Failed to refresh datamap $dataMapName on segment_$segmentId")
-          }
-        } catch {
-          case ex: Throwable =>
-            // process failure
-            FileFactory.deleteAllCarbonFilesOfDir(FileFactory.getCarbonFile(dataMapStorePath))
-            throw new Exception(
-              s"Failed to refresh datamap $dataMapName on segment_$segmentId", ex)
-        }
-      } else {
-        throw new IOException(s"Failed to create directory $dataMapStorePath")
-      }
-    }
-  }
-
-}
-
-class OriginalReadSupport(dataTypes: Array[DataType]) extends CarbonReadSupport[Array[Object]] {
-  override def initialize(carbonColumns: Array[CarbonColumn],
-      carbonTable: CarbonTable): Unit = {
-  }
-
-  override def readRow(data: Array[Object]): Array[Object] = {
-    dataTypes.zipWithIndex.foreach { case (dataType, i) =>
-      if (dataType == DataTypes.STRING) {
-        data(i) = data(i).toString
-      }
-    }
-    data
-  }
-
-  override def close(): Unit = {
-  }
-}
-
-class IndexDataMapRefreshRDD[K, V](
-    session: SparkSession,
-    result: RefreshResult[K, V],
-    @transient tableInfo: TableInfo,
-    dataMapName: String,
-    indexColumns: Array[CarbonColumn],
-    segmentId: String
-) extends CarbonRDDWithTableInfo[(K, V)](
-  session.sparkContext, Nil, tableInfo.serialize()) {
-
-  private val dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
-  private val queryId = sparkContext.getConf.get("queryId", System.nanoTime() + "")
-  private val jobTrackerId: String = {
-    val formatter = new SimpleDateFormat("yyyyMMddHHmm")
-    formatter.format(new util.Date())
-  }
-
-  override def internalCompute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
-    val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
-    val dataMapFactory =
-      DataMapManager.get().getDataMapProvider(
-        CarbonTable.buildFromTableInfo(getTableInfo), dataMapSchema, session).getDataMapFactory
-    var status = false
-    val inputMetrics = new CarbonInputMetrics
-    TaskMetricsMap.getInstance().registerThreadCallback()
-    val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value
-    inputMetrics.initBytesReadCallback(context, inputSplit)
-
-    val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
-    val attemptContext = new TaskAttemptContextImpl(new Configuration(), attemptId)
-    val format = createInputFormat(attemptContext)
-
-    val model = format.createQueryModel(inputSplit, attemptContext)
-    // one query id per table
-    model.setQueryId(queryId)
-    model.setVectorReader(false)
-    model.setForcedDetailRawQuery(false)
-    model.setRequiredRowId(true)
-
-    var reader: CarbonRecordReader[Array[Object]] = null
-    var refresher: DataMapRefresher = null
-    try {
-      reader = new CarbonRecordReader(
-        model, new OriginalReadSupport(indexColumns.map(_.getDataType)), inputMetrics)
-      reader.initialize(inputSplit, attemptContext)
-
-      // we use task name as shard name to create the folder for this datamap
-      val shardName = CarbonTablePath.getShardName(inputSplit.getAllSplits.get(0).getBlockPath)
-      refresher = dataMapFactory.createRefresher(new Segment(segmentId), shardName)
-      refresher.initialize()
-
-      var blockletId = 0
-      var firstRow = true
-      while (reader.nextKeyValue()) {
-        val rowWithPosition = reader.getCurrentValue
-        val size = rowWithPosition.length
-        val pageId = rowWithPosition(size - 2).asInstanceOf[Int]
-        val rowId = rowWithPosition(size - 1).asInstanceOf[Int]
-
-        if (!firstRow && pageId == 0 && rowId == 0) {
-          // new blocklet started, increase blockletId
-          blockletId = blockletId + 1
-        } else {
-          firstRow = false
-        }
-
-        refresher.addRow(blockletId, pageId, rowId, rowWithPosition)
-      }
-
-      refresher.finish()
-
-      status = true
-    } finally {
-      if (reader != null) {
-        try {
-          reader.close()
-        } catch {
-          case ex: Throwable =>
-            LOGGER.error(ex, "Failed to close reader")
-        }
-      }
-
-      if (refresher != null) {
-        try {
-          refresher.close()
-        } catch {
-          case ex: Throwable =>
-            LOGGER.error(ex, "Failed to close index writer")
-        }
-      }
-    }
-
-    new Iterator[(K, V)] {
-
-      var finished = false
-
-      override def hasNext: Boolean = {
-        !finished
-      }
-
-      override def next(): (K, V) = {
-        finished = true
-        result.getKey(split.index.toString, status)
-      }
-    }
-  }
-
-
-  private def createInputFormat(
-      attemptContext: TaskAttemptContextImpl) = {
-    val format = new CarbonTableInputFormat[Object]
-    val tableInfo1 = getTableInfo
-    val conf = attemptContext.getConfiguration
-    CarbonInputFormat.setTableInfo(conf, tableInfo1)
-    CarbonInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)
-    CarbonInputFormat.setTableName(conf, tableInfo1.getFactTable.getTableName)
-    CarbonInputFormat.setDataTypeConverter(conf, classOf[SparkDataTypeConverterImpl])
-
-    val identifier = tableInfo1.getOrCreateAbsoluteTableIdentifier()
-    CarbonInputFormat.setTablePath(
-      conf,
-      identifier.appendWithLocalPrefix(identifier.getTablePath))
-
-    CarbonInputFormat.setSegmentsToAccess(
-      conf,
-      Segment.toSegmentList(Array(segmentId), null))
-
-    CarbonInputFormat.setColumnProjection(
-      conf,
-      new CarbonProjection(indexColumns.map(_.getColName)))
-    format
-  }
-
-  override protected def getPartitions = {
-    if (!dataMapSchema.isIndexDataMap) {
-      throw new UnsupportedOperationException
-    }
-    val conf = new Configuration()
-    val jobConf = new JobConf(conf)
-    SparkHadoopUtil.get.addCredentials(jobConf)
-    val job = Job.getInstance(jobConf)
-    job.getConfiguration.set("query.id", queryId)
-
-    val format = new CarbonTableInputFormat[Object]
-
-    CarbonInputFormat.setSegmentsToAccess(
-      job.getConfiguration,
-      Segment.toSegmentList(Array(segmentId), null))
-
-    CarbonInputFormat.setTableInfo(
-      job.getConfiguration,
-      tableInfo)
-    CarbonInputFormat.setTablePath(
-      job.getConfiguration,
-      tableInfo.getOrCreateAbsoluteTableIdentifier().getTablePath)
-    CarbonInputFormat.setDatabaseName(
-      job.getConfiguration,
-      tableInfo.getDatabaseName)
-    CarbonInputFormat.setTableName(
-      job.getConfiguration,
-      tableInfo.getFactTable.getTableName)
-
-    format
-      .getSplits(job)
-      .asScala
-      .map(_.asInstanceOf[CarbonInputSplit])
-      .groupBy(_.taskId)
-      .map { group =>
-        new CarbonMultiBlockSplit(
-          group._2.asJava,
-          group._2.flatMap(_.getLocations).toArray)
-      }
-      .zipWithIndex
-      .map { split =>
-        new CarbonSparkPartition(id, split._2, split._1)
-      }
-      .toArray
-  }
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 42c9c25..bdbaef5 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -848,11 +848,11 @@ object CarbonDataRDDFactory {
       val errorMessage = s"Dataload failed due to failure in table status updation for" +
                          s" ${carbonLoadModel.getTableName}"
       LOGGER.audit("Data load is failed for " +
-                   s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+                   s"${carbonLoadModel.getDatabaseName}.${carbonLoadModel.getTableName}")
       LOGGER.error("Dataload failed due to failure in table status updation.")
       throw new Exception(errorMessage)
     } else {
-      DataMapStatusManager.disableDataMapsOfTable(carbonTable)
+      DataMapStatusManager.disableAllLazyDataMaps(carbonTable)
     }
     done
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
index 5fac5a8..497f95a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
@@ -36,6 +36,7 @@ import org.apache.spark.sql.optimizer.CarbonFilters
 import org.apache.spark.sql.profiler.{Profiler, SQLStart}
 import org.apache.spark.util.{CarbonReflectionUtils, Utils}
 
+import org.apache.carbondata.common.annotations.InterfaceAudience
 import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.scan.expression.LiteralExpression
@@ -110,6 +111,16 @@ class CarbonSession(@transient val sc: SparkContext,
     )
   }
 
+  /**
+   * Return true if the specified sql statement will hit the datamap
+   * This API is for test purpose only
+   */
+  @InterfaceAudience.Developer(Array("DataMap"))
+  def isDataMapHit(sqlStatement: String, dataMapName: String): Boolean = {
+    val message = sql(s"EXPLAIN $sqlStatement").collect()
+    message(0).getString(0).contains(dataMapName)
+  }
+
   def isSearchModeEnabled: Boolean = carbonStore != null
 
   /**

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
index 34a4013..25589d4 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql.execution.command.datamap
 
+import java.util
+
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 
@@ -27,7 +29,7 @@ import org.apache.carbondata.common.exceptions.sql.{MalformedCarbonCommandExcept
 import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.datamap.{DataMapProvider, DataMapStoreManager}
 import org.apache.carbondata.core.datamap.status.DataMapStatusManager
-import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
+import org.apache.carbondata.core.metadata.schema.datamap.{DataMapClassProvider, DataMapProperty}
 import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
 import org.apache.carbondata.datamap.{DataMapManager, IndexDataMapProvider}
 
@@ -41,7 +43,8 @@ case class CarbonCreateDataMapCommand(
     dmProviderName: String,
     dmProperties: Map[String, String],
     queryString: Option[String],
-    ifNotExistsSet: Boolean = false)
+    ifNotExistsSet: Boolean = false,
+    deferredRebuild: Boolean = false)
   extends AtomicRunnableCommand {
 
   private var dataMapProvider: DataMapProvider = _
@@ -78,9 +81,16 @@ case class CarbonCreateDataMapCommand(
     }
 
     dataMapSchema = new DataMapSchema(dataMapName, dmProviderName)
-    dataMapSchema.setProperties(new java.util.HashMap[String, String](
-      dmProperties.map(x => (x._1.trim, x._2.trim)).asJava))
 
+    val property = dmProperties.map(x => (x._1.trim, x._2.trim)).asJava
+    val javaMap = new java.util.HashMap[String, String](property)
+    javaMap.put(DataMapProperty.DEFERRED_REBUILD, deferredRebuild.toString)
+    dataMapSchema.setProperties(javaMap)
+
+    if (dataMapSchema.isIndexDataMap && mainTable == null) {
+      throw new MalformedDataMapCommandException(
+        "For this datamap, main table is required. Use `CREATE DATAMAP ... ON TABLE ...` ")
+    }
     dataMapProvider = DataMapManager.get.getDataMapProvider(mainTable, dataMapSchema, sparkSession)
 
     // If it is index datamap, check whether the column has datamap created already
@@ -101,6 +111,10 @@ case class CarbonCreateDataMapCommand(
         dataMapProvider.initMeta(queryString.orNull)
         DataMapStatusManager.disableDataMap(dataMapName)
       case _ =>
+        if (deferredRebuild) {
+          throw new MalformedDataMapCommandException(
+            "DEFERRED REBUILD is not supported on this DataMap")
+        }
         dataMapProvider.initMeta(queryString.orNull)
     }
     val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
@@ -111,10 +125,11 @@ case class CarbonCreateDataMapCommand(
   override def processData(sparkSession: SparkSession): Seq[Row] = {
     if (dataMapProvider != null) {
       dataMapProvider.initData()
-      if (mainTable != null &&
-          mainTable.isAutoRefreshDataMap &&
-          !dataMapSchema.isIndexDataMap) {
+      if (mainTable != null && !deferredRebuild) {
         dataMapProvider.rebuild()
+        if (dataMapSchema.isIndexDataMap) {
+          DataMapStatusManager.enableDataMap(dataMapName)
+        }
       }
     }
     Seq.empty

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRebuildCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRebuildCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRebuildCommand.scala
new file mode 100644
index 0000000..6493c83
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRebuildCommand.scala
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command.datamap
+
+import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.execution.command.DataCommand
+
+import org.apache.carbondata.core.datamap.{DataMapRegistry, DataMapStoreManager}
+import org.apache.carbondata.core.datamap.status.DataMapStatusManager
+import org.apache.carbondata.datamap.{DataMapManager, IndexDataMapRebuildRDD}
+
+/**
+ * Rebuild the datamaps through sync with main table data. After sync with parent table's it enables
+ * the datamap.
+ */
+case class CarbonDataMapRebuildCommand(
+    dataMapName: String,
+    tableIdentifier: Option[TableIdentifier]) extends DataCommand {
+
+  override def processData(sparkSession: SparkSession): Seq[Row] = {
+    val schema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
+
+    val table = tableIdentifier match {
+      case Some(identifier) =>
+        CarbonEnv.getCarbonTable(identifier)(sparkSession)
+      case _ =>
+        CarbonEnv.getCarbonTable(
+          Option(schema.getRelationIdentifier.getDatabaseName),
+          schema.getRelationIdentifier.getTableName
+        )(sparkSession)
+    }
+    val provider = DataMapManager.get().getDataMapProvider(table, schema, sparkSession)
+    provider.rebuild()
+
+    // After rebuild successfully enable the datamap.
+    DataMapStatusManager.enableDataMap(dataMapName)
+    Seq.empty
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
deleted file mode 100644
index 4f3b7bc..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapRefreshCommand.scala
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.command.datamap
-
-import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
-import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.execution.command.DataCommand
-
-import org.apache.carbondata.core.datamap.{DataMapRegistry, DataMapStoreManager}
-import org.apache.carbondata.core.datamap.status.DataMapStatusManager
-import org.apache.carbondata.datamap.{DataMapManager, IndexDataMapRefreshRDD}
-
-/**
- * Refresh the datamaps through sync with main table data. After sync with parent table's it enables
- * the datamap.
- */
-case class CarbonDataMapRefreshCommand(
-    dataMapName: String,
-    tableIdentifier: Option[TableIdentifier]) extends DataCommand {
-
-  override def processData(sparkSession: SparkSession): Seq[Row] = {
-    val schema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
-
-    val table = tableIdentifier match {
-      case Some(identifier) =>
-        CarbonEnv.getCarbonTable(identifier)(sparkSession)
-      case _ =>
-        CarbonEnv.getCarbonTable(
-          Option(schema.getRelationIdentifier.getDatabaseName),
-          schema.getRelationIdentifier.getTableName
-        )(sparkSession)
-    }
-    val provider = DataMapManager.get().getDataMapProvider(table, schema, sparkSession)
-    provider.rebuild()
-
-    // After sync success enable the datamap.
-    DataMapStatusManager.enableDataMap(dataMapName)
-    Seq.empty
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala
index 1bb7d7c..4f60297 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonFileFormat.scala
@@ -231,8 +231,9 @@ class SparkCarbonFileFormat extends FileFormat
 
         val tab = model.getTable
         DataMapStoreManager.getInstance().clearDataMaps(identifier)
-        val dataMapExprWrapper = DataMapChooser.get
-          .choose(tab, model.getFilterExpressionResolverTree)
+
+        val dataMapExprWrapper = new DataMapChooser(tab).choose(
+          model.getFilterExpressionResolverTree)
 
         // TODO : handle the partition for CarbonFileLevelFormat
         val prunedBlocklets = dataMapExprWrapper.prune(segments, null)

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index 62da7ed..9dd8105 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.{CarbonDDLSqlParser, TableIdentifier}
 import org.apache.spark.sql.catalyst.CarbonTableIdentifierImplicit._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.execution.command._
-import org.apache.spark.sql.execution.command.datamap.{CarbonCreateDataMapCommand, CarbonDataMapRefreshCommand, CarbonDataMapShowCommand, CarbonDropDataMapCommand}
+import org.apache.spark.sql.execution.command.datamap.{CarbonCreateDataMapCommand, CarbonDataMapRebuildCommand, CarbonDataMapShowCommand, CarbonDropDataMapCommand}
 import org.apache.spark.sql.execution.command.management._
 import org.apache.spark.sql.execution.command.partition.{CarbonAlterTableDropPartitionCommand, CarbonAlterTableSplitPartitionCommand}
 import org.apache.spark.sql.execution.command.schema.{CarbonAlterTableAddColumnCommand, CarbonAlterTableDataTypeChangeCommand, CarbonAlterTableDropColumnCommand}
@@ -147,19 +147,23 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
 
   /**
    * The syntax of datamap creation is as follows.
-   * CREATE DATAMAP IF NOT EXISTS datamapName ON TABLE tableName USING 'DataMapProviderName'
+   * CREATE DATAMAP IF NOT EXISTS datamapName [ON TABLE tableName]
+   * USING 'DataMapProviderName'
+   * [WITH DEFERRED REBUILD]
    * DMPROPERTIES('KEY'='VALUE') AS SELECT COUNT(COL1) FROM tableName
    */
   protected lazy val createDataMap: Parser[LogicalPlan] =
     CREATE ~> DATAMAP ~> opt(IF ~> NOT ~> EXISTS) ~ ident ~
     opt(ontable) ~
-    (USING ~> stringLit) ~ (DMPROPERTIES ~> "(" ~> repsep(loadOptions, ",") <~ ")").? ~
+    (USING ~> stringLit) ~
+    opt(WITH ~> DEFERRED ~> REBUILD) ~
+    (DMPROPERTIES ~> "(" ~> repsep(loadOptions, ",") <~ ")").? ~
     (AS ~> restInput).? <~ opt(";") ^^ {
-      case ifnotexists ~ dmname ~ tableIdent ~ dmProviderName ~ dmprops ~ query =>
+      case ifnotexists ~ dmname ~ tableIdent ~ dmProviderName ~ deferred ~ dmprops ~ query =>
 
         val map = dmprops.getOrElse(List[(String, String)]()).toMap[String, String]
         CarbonCreateDataMapCommand(dmname, tableIdent, dmProviderName, map, query,
-          ifnotexists.isDefined)
+          ifnotexists.isDefined, deferred.isDefined)
     }
 
   protected lazy val ontable: Parser[TableIdentifier] =
@@ -190,12 +194,12 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
 
   /**
    * The syntax of show datamap is used to show datamaps on the table
-   * REFRESH DATAMAP datamapname [ON TABLE] tableName
+   * REBUILD DATAMAP datamapname [ON TABLE] tableName
    */
   protected lazy val refreshDataMap: Parser[LogicalPlan] =
-    REFRESH ~> DATAMAP ~> ident ~ opt(ontable) <~ opt(";") ^^ {
+    REBUILD ~> DATAMAP ~> ident ~ opt(ontable) <~ opt(";") ^^ {
       case datamap ~ tableIdent =>
-        CarbonDataMapRefreshCommand(datamap, tableIdent)
+        CarbonDataMapRebuildCommand(datamap, tableIdent)
     }
 
   protected lazy val deleteRecords: Parser[LogicalPlan] =

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala b/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
index a8e7b6c..7df3901 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapSuite.scala
@@ -22,44 +22,56 @@ import java.util.UUID
 
 import scala.util.Random
 
+import org.apache.spark.sql.{CarbonSession, DataFrame}
 import org.apache.spark.sql.test.util.QueryTest
-import org.scalatest.BeforeAndAfterAll
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 
+import org.apache.carbondata.core.datamap.status.DataMapStatusManager
 import org.apache.carbondata.core.util.CarbonProperties
 
-class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
-  val inputFile = s"$resourcesPath/bloom_datamap_input.csv"
+class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll with BeforeAndAfterEach {
+  val bigFile = s"$resourcesPath/bloom_datamap_input_big.csv"
+  val smallFile = s"$resourcesPath/bloom_datamap_input_small.csv"
   val normalTable = "carbon_normal"
   val bloomDMSampleTable = "carbon_bloom"
   val dataMapName = "bloom_dm"
-  val lineNum = 500000
 
   override protected def beforeAll(): Unit = {
     new File(CarbonProperties.getInstance().getSystemFolderLocation).delete()
-    createFile(inputFile, line = lineNum, start = 0)
+    createFile(bigFile, line = 500000)
+    createFile(smallFile)
     sql(s"DROP TABLE IF EXISTS $normalTable")
     sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
   }
 
-  private def checkQuery = {
+  override def afterEach(): Unit = {
+    sql(s"DROP TABLE IF EXISTS $normalTable")
+    sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
+  }
+
+  private def checkSqlHitDataMap(sqlText: String, dataMapName: String, shouldHit: Boolean): DataFrame = {
+    if (shouldHit) {
+      assert(sqlContext.sparkSession.asInstanceOf[CarbonSession].isDataMapHit(sqlText, dataMapName))
+    } else {
+      assert(!sqlContext.sparkSession.asInstanceOf[CarbonSession].isDataMapHit(sqlText, dataMapName))
+    }
+    sql(sqlText)
+  }
+
+  private def checkQuery(dataMapName: String, shouldHit: Boolean = true) = {
     checkAnswer(
-      sql(s"select * from $bloomDMSampleTable where id = 1"),
+      checkSqlHitDataMap(s"select * from $bloomDMSampleTable where id = 1", dataMapName, shouldHit),
       sql(s"select * from $normalTable where id = 1"))
     checkAnswer(
-      sql(s"select * from $bloomDMSampleTable where id = 999"),
+      checkSqlHitDataMap(s"select * from $bloomDMSampleTable where id = 999", dataMapName, shouldHit),
       sql(s"select * from $normalTable where id = 999"))
     checkAnswer(
-      sql(s"select * from $bloomDMSampleTable where city = 'city_1'"),
+      checkSqlHitDataMap(s"select * from $bloomDMSampleTable where city = 'city_1'", dataMapName, shouldHit),
       sql(s"select * from $normalTable where city = 'city_1'"))
     checkAnswer(
-      sql(s"select * from $bloomDMSampleTable where city = 'city_999'"),
+      checkSqlHitDataMap(s"select * from $bloomDMSampleTable where city = 'city_999'", dataMapName, shouldHit),
       sql(s"select * from $normalTable where city = 'city_999'"))
-    checkAnswer(
-      sql(s"select count(distinct id), count(distinct name), count(distinct city)," +
-          s" count(distinct s1), count(distinct s2) from $bloomDMSampleTable"),
-      sql(s"select count(distinct id), count(distinct name), count(distinct city)," +
-          s" count(distinct s1), count(distinct s2) from $normalTable"))
-    checkAnswer(
+     checkAnswer(
       sql(s"select min(id), max(id), min(name), max(name), min(city), max(city)" +
           s" from $bloomDMSampleTable"),
       sql(s"select min(id), max(id), min(name), max(name), min(city), max(city)" +
@@ -86,28 +98,34 @@ class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
          | DMProperties('INDEX_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
       """.stripMargin)
 
+    var map = DataMapStatusManager.readDataMapStatusMap()
+    assert(map.get(dataMapName).isEnabled)
+
     // load two segments
     (1 to 2).foreach { i =>
       sql(
         s"""
-           | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $normalTable
+           | LOAD DATA LOCAL INPATH '$bigFile' INTO TABLE $normalTable
            | OPTIONS('header'='false')
          """.stripMargin)
       sql(
         s"""
-           | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $bloomDMSampleTable
+           | LOAD DATA LOCAL INPATH '$bigFile' INTO TABLE $bloomDMSampleTable
            | OPTIONS('header'='false')
          """.stripMargin)
     }
 
+    map = DataMapStatusManager.readDataMapStatusMap()
+    assert(map.get(dataMapName).isEnabled)
+
     sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable").show(false)
     checkExistence(sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable"), true, dataMapName)
-    checkQuery
+    checkQuery(dataMapName)
     sql(s"DROP TABLE IF EXISTS $normalTable")
     sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
   }
 
-  test("test create bloom datamap and refresh datamap") {
+  test("test create bloom datamap and REBUILD DATAMAP") {
     sql(
       s"""
          | CREATE TABLE $normalTable(id INT, name STRING, city STRING, age INT,
@@ -125,12 +143,12 @@ class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
     (1 to 2).foreach { i =>
       sql(
         s"""
-           | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $normalTable
+           | LOAD DATA LOCAL INPATH '$bigFile' INTO TABLE $normalTable
            | OPTIONS('header'='false')
          """.stripMargin)
       sql(
         s"""
-           | LOAD DATA LOCAL INPATH '$inputFile' INTO TABLE $bloomDMSampleTable
+           | LOAD DATA LOCAL INPATH '$bigFile' INTO TABLE $bloomDMSampleTable
            | OPTIONS('header'='false')
          """.stripMargin)
     }
@@ -142,18 +160,135 @@ class BloomCoarseGrainDataMapSuite extends QueryTest with BeforeAndAfterAll {
          | DMProperties('INDEX_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
       """.stripMargin)
 
-    sql(s"REFRESH DATAMAP $dataMapName ON TABLE $bloomDMSampleTable")
     sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable").show(false)
     checkExistence(sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable"), true, dataMapName)
-    checkQuery
+    checkQuery(dataMapName)
+    sql(s"DROP TABLE IF EXISTS $normalTable")
+    sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
+  }
+
+  test("test create bloom datamap with DEFERRED REBUILD, query hit datamap") {
+    sql(
+      s"""
+         | CREATE TABLE $normalTable(id INT, name STRING, city STRING, age INT,
+         | s1 STRING, s2 STRING, s3 STRING, s4 STRING, s5 STRING, s6 STRING, s7 STRING, s8 STRING)
+         | STORED BY 'carbondata' TBLPROPERTIES('table_blocksize'='128')
+         |  """.stripMargin)
+    sql(
+      s"""
+         | CREATE TABLE $bloomDMSampleTable(id INT, name STRING, city STRING, age INT,
+         | s1 STRING, s2 STRING, s3 STRING, s4 STRING, s5 STRING, s6 STRING, s7 STRING, s8 STRING)
+         | STORED BY 'carbondata' TBLPROPERTIES('table_blocksize'='128')
+         |  """.stripMargin)
+
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $normalTable
+         | OPTIONS('header'='false')
+       """.stripMargin)
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $bloomDMSampleTable
+         | OPTIONS('header'='false')
+       """.stripMargin)
+
+    sql(
+      s"""
+         | CREATE DATAMAP $dataMapName ON TABLE $bloomDMSampleTable
+         | USING 'bloomfilter'
+         | WITH DEFERRED REBUILD
+         | DMProperties('INDEX_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
+      """.stripMargin)
+
+    var map = DataMapStatusManager.readDataMapStatusMap()
+    assert(!map.get(dataMapName).isEnabled)
+
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $normalTable
+         | OPTIONS('header'='false')
+         """.stripMargin)
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $bloomDMSampleTable
+         | OPTIONS('header'='false')
+         """.stripMargin)
+
+    map = DataMapStatusManager.readDataMapStatusMap()
+    assert(!map.get(dataMapName).isEnabled)
+
+    // once we rebuild, it should be enabled
+    sql(s"REBUILD DATAMAP $dataMapName ON TABLE $bloomDMSampleTable")
+    map = DataMapStatusManager.readDataMapStatusMap()
+    assert(map.get(dataMapName).isEnabled)
+
+    sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable").show(false)
+    checkExistence(sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable"), true, dataMapName)
+    checkQuery(dataMapName)
+
+    // once we load again, datamap should be disabled, since it is lazy
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $normalTable
+         | OPTIONS('header'='false')
+         """.stripMargin)
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $bloomDMSampleTable
+         | OPTIONS('header'='false')
+         """.stripMargin)
+    map = DataMapStatusManager.readDataMapStatusMap()
+    assert(!map.get(dataMapName).isEnabled)
+    checkQuery(dataMapName, shouldHit = false)
+
     sql(s"DROP TABLE IF EXISTS $normalTable")
     sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
   }
 
-  // todo: will add more tests on bloom datamap, such as exception, delete datamap, show profiler
+  test("test create bloom datamap with DEFERRED REBUILD, query not hit datamap") {
+    sql(
+      s"""
+         | CREATE TABLE $normalTable(id INT, name STRING, city STRING, age INT,
+         | s1 STRING, s2 STRING, s3 STRING, s4 STRING, s5 STRING, s6 STRING, s7 STRING, s8 STRING)
+         | STORED BY 'carbondata' TBLPROPERTIES('table_blocksize'='128')
+         |  """.stripMargin)
+    sql(
+      s"""
+         | CREATE TABLE $bloomDMSampleTable(id INT, name STRING, city STRING, age INT,
+         | s1 STRING, s2 STRING, s3 STRING, s4 STRING, s5 STRING, s6 STRING, s7 STRING, s8 STRING)
+         | STORED BY 'carbondata' TBLPROPERTIES('table_blocksize'='128')
+         |  """.stripMargin)
+
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $normalTable
+         | OPTIONS('header'='false')
+       """.stripMargin)
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '$smallFile' INTO TABLE $bloomDMSampleTable
+         | OPTIONS('header'='false')
+       """.stripMargin)
+
+    sql(
+      s"""
+         | CREATE DATAMAP $dataMapName ON TABLE $bloomDMSampleTable
+         | USING 'bloomfilter'
+         | WITH DEFERRED REBUILD
+         | DMProperties('INDEX_COLUMNS'='city,id', 'BLOOM_SIZE'='640000')
+      """.stripMargin)
+
+    checkExistence(sql(s"SHOW DATAMAP ON TABLE $bloomDMSampleTable"), true, dataMapName)
+
+    // datamap is not loaded, so it should not hit
+    checkQuery(dataMapName, shouldHit = false)
+    sql(s"DROP TABLE IF EXISTS $normalTable")
+    sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
+  }
 
   override protected def afterAll(): Unit = {
-    deleteFile(inputFile)
+    deleteFile(bigFile)
+    deleteFile(smallFile)
     sql(s"DROP TABLE IF EXISTS $normalTable")
     sql(s"DROP TABLE IF EXISTS $bloomDMSampleTable")
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java b/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
index 9c3d5d6..3dc34d3 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/datamap/DataMapWriterListener.java
@@ -61,8 +61,12 @@ public class DataMapWriterListener {
     }
     if (tableIndices != null) {
       for (TableDataMap tableDataMap : tableIndices) {
-        DataMapFactory factory = tableDataMap.getDataMapFactory();
-        register(factory, segmentId, taskNo);
+        // register it only if it is not lazy datamap, for lazy datamap, user
+        // will rebuild the datamap manually
+        if (!tableDataMap.getDataMapSchema().isLazy()) {
+          DataMapFactory factory = tableDataMap.getDataMapFactory();
+          register(factory, segmentId, taskNo);
+        }
       }
     }
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/store/search/src/main/java/org/apache/carbondata/store/worker/SearchRequestHandler.java
----------------------------------------------------------------------
diff --git a/store/search/src/main/java/org/apache/carbondata/store/worker/SearchRequestHandler.java b/store/search/src/main/java/org/apache/carbondata/store/worker/SearchRequestHandler.java
index 957e9f8..35acb17 100644
--- a/store/search/src/main/java/org/apache/carbondata/store/worker/SearchRequestHandler.java
+++ b/store/search/src/main/java/org/apache/carbondata/store/worker/SearchRequestHandler.java
@@ -21,12 +21,11 @@ import java.io.IOException;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Objects;
 
 import org.apache.carbondata.common.annotations.InterfaceAudience;
 import org.apache.carbondata.common.logging.LogService;
 import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.datamap.DataMapChooser;
-import org.apache.carbondata.core.datamap.DataMapLevel;
 import org.apache.carbondata.core.datamap.Segment;
 import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper;
 import org.apache.carbondata.core.datastore.block.TableBlockInfo;
@@ -104,8 +103,10 @@ public class SearchRequestHandler {
     LOG.info(String.format("[SearchId:%d] %s, number of block: %d",
         request.searchId(), queryModel.toString(), mbSplit.getAllSplits().size()));
 
-    // If there is FGDataMap, prune the split by applying FGDataMap
-    queryModel = tryPruneByFGDataMap(request.searchId(), table, queryModel, mbSplit);
+    // If there is DataMap selected in Master, prune the split by it
+    if (request.dataMap() != null) {
+      queryModel = prune(request.searchId(), table, queryModel, mbSplit, request.dataMap().get());
+    }
 
     // In search mode, reader will read multiple blocks by using a thread pool
     CarbonRecordReader<CarbonRow> reader =
@@ -135,35 +136,32 @@ public class SearchRequestHandler {
    * If there is FGDataMap defined for this table and filter condition in the query,
    * prune the splits by the DataMap and set the pruned split into the QueryModel and return
    */
-  private QueryModel tryPruneByFGDataMap(int queryId,
-      CarbonTable table, QueryModel queryModel, CarbonMultiBlockSplit mbSplit) throws IOException {
-    DataMapExprWrapper wrapper =
-        DataMapChooser.get().choose(table, queryModel.getFilterExpressionResolverTree());
-
-    if (wrapper.getDataMapLevel() == DataMapLevel.FG) {
-      List<Segment> segments = new LinkedList<>();
-      for (CarbonInputSplit split : mbSplit.getAllSplits()) {
-        segments.add(Segment.toSegment(
-            split.getSegmentId(), new LatestFilesReadCommittedScope(table.getTablePath())));
-      }
-      List<ExtendedBlocklet> prunnedBlocklets = wrapper.prune(segments, null);
+  private QueryModel prune(int queryId, CarbonTable table, QueryModel queryModel,
+      CarbonMultiBlockSplit mbSplit, DataMapExprWrapper datamap) throws IOException {
+    Objects.requireNonNull(datamap);
+    List<Segment> segments = new LinkedList<>();
+    for (CarbonInputSplit split : mbSplit.getAllSplits()) {
+      segments.add(
+          Segment.toSegment(split.getSegmentId(),
+              new LatestFilesReadCommittedScope(table.getTablePath())));
+    }
+    List<ExtendedBlocklet> prunnedBlocklets = datamap.prune(segments, null);
 
-      List<String> pathToRead = new LinkedList<>();
-      for (ExtendedBlocklet prunnedBlocklet : prunnedBlocklets) {
-        pathToRead.add(prunnedBlocklet.getPath());
-      }
+    List<String> pathToRead = new LinkedList<>();
+    for (ExtendedBlocklet prunnedBlocklet : prunnedBlocklets) {
+      pathToRead.add(prunnedBlocklet.getPath());
+    }
 
-      List<TableBlockInfo> blocks = queryModel.getTableBlockInfos();
-      List<TableBlockInfo> blockToRead = new LinkedList<>();
-      for (TableBlockInfo block : blocks) {
-        if (pathToRead.contains(block.getFilePath())) {
-          blockToRead.add(block);
-        }
+    List<TableBlockInfo> blocks = queryModel.getTableBlockInfos();
+    List<TableBlockInfo> blockToRead = new LinkedList<>();
+    for (TableBlockInfo block : blocks) {
+      if (pathToRead.contains(block.getFilePath())) {
+        blockToRead.add(block);
       }
-      LOG.info(String.format("[SearchId:%d] pruned using FG DataMap, pruned blocks: %d",
-          queryId, blockToRead.size()));
-      queryModel.setTableBlockInfos(blockToRead);
     }
+    LOG.info(String.format("[SearchId:%d] pruned using FG DataMap, pruned blocks: %d", queryId,
+        blockToRead.size()));
+    queryModel.setTableBlockInfos(blockToRead);
     return queryModel;
   }
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/store/search/src/main/scala/org/apache/spark/rpc/Master.scala
----------------------------------------------------------------------
diff --git a/store/search/src/main/scala/org/apache/spark/rpc/Master.scala b/store/search/src/main/scala/org/apache/spark/rpc/Master.scala
index 5b31a49..2e9a532 100644
--- a/store/search/src/main/scala/org/apache/spark/rpc/Master.scala
+++ b/store/search/src/main/scala/org/apache/spark/rpc/Master.scala
@@ -38,12 +38,15 @@ import org.apache.spark.util.ThreadUtils
 
 import org.apache.carbondata.common.annotations.InterfaceAudience
 import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.datamap.DataMapChooser
+import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper
 import org.apache.carbondata.core.datastore.block.Distributable
 import org.apache.carbondata.core.datastore.row.CarbonRow
 import org.apache.carbondata.core.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.scan.expression.Expression
 import org.apache.carbondata.core.util.CarbonProperties
 import org.apache.carbondata.hadoop.CarbonMultiBlockSplit
+import org.apache.carbondata.hadoop.api.CarbonInputFormat
 import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
 import org.apache.carbondata.processing.util.CarbonLoaderUtil
 import org.apache.carbondata.store.worker.Status
@@ -212,11 +215,13 @@ class Master(sparkConf: SparkConf) {
     // prune data and get a mapping of worker hostname to list of blocks,
     // then add these blocks to the SearchRequest and fire the RPC call
     val nodeBlockMapping: JMap[String, JList[Distributable]] = pruneBlock(table, columns, filter)
+    val fgDataMap = chooseFGDataMap(table, filter)
     val tuple = nodeBlockMapping.asScala.map { case (splitAddress, blocks) =>
       // Build a SearchRequest
       val split = new SerializableWritable[CarbonMultiBlockSplit](
         new CarbonMultiBlockSplit(blocks, splitAddress))
-      val request = SearchRequest(queryId, split, table.getTableInfo, columns, filter, localLimit)
+      val request =
+        SearchRequest(queryId, split, table.getTableInfo, columns, filter, localLimit, fgDataMap)
 
       // Find an Endpoind and send the request to it
       // This RPC is non-blocking so that we do not need to wait before send to next worker
@@ -249,6 +254,14 @@ class Master(sparkConf: SparkConf) {
     output.toArray
   }
 
+  private def chooseFGDataMap(
+      table: CarbonTable,
+      filter: Expression): Option[DataMapExprWrapper] = {
+    val chooser = new DataMapChooser(table)
+    val filterInterface = table.resolveFilter(filter)
+    Option(chooser.chooseFGDataMap(filterInterface))
+  }
+
   /**
    * Prune data by using CarbonInputFormat.getSplit
    * Return a mapping of host address to list of block
@@ -261,6 +274,9 @@ class Master(sparkConf: SparkConf) {
     val job = new Job(jobConf)
     val format = CarbonInputFormatUtil.createCarbonTableInputFormat(
       job, table, columns, filter, null, null)
+
+    // We will do FG pruning in reader side, so don't do it here
+    CarbonInputFormat.setFgDataMapPruning(job.getConfiguration, false)
     val splits = format.getSplits(job)
     val distributables = splits.asScala.map { split =>
       split.asInstanceOf[Distributable]

http://git-wip-us.apache.org/repos/asf/carbondata/blob/747be9b1/store/search/src/main/scala/org/apache/spark/search/Searcher.scala
----------------------------------------------------------------------
diff --git a/store/search/src/main/scala/org/apache/spark/search/Searcher.scala b/store/search/src/main/scala/org/apache/spark/search/Searcher.scala
index e467fd3..1532284 100644
--- a/store/search/src/main/scala/org/apache/spark/search/Searcher.scala
+++ b/store/search/src/main/scala/org/apache/spark/search/Searcher.scala
@@ -21,6 +21,7 @@ import org.apache.spark.SerializableWritable
 import org.apache.spark.rpc.{RpcCallContext, RpcEndpoint, RpcEnv}
 
 import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.datamap.dev.expr.DataMapExprWrapper
 import org.apache.carbondata.core.metadata.schema.table.TableInfo
 import org.apache.carbondata.core.scan.expression.Expression
 import org.apache.carbondata.hadoop.CarbonMultiBlockSplit
@@ -37,11 +38,11 @@ class Searcher(override val rpcEnv: RpcEnv) extends RpcEndpoint {
   }
 
   override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
-    case req@SearchRequest(_, _, _, _, _, _) =>
+    case req: SearchRequest =>
       val response = new SearchRequestHandler().handleSearch(req)
       context.reply(response)
 
-    case req@ShutdownRequest(_) =>
+    case req: ShutdownRequest =>
       val response = new SearchRequestHandler().handleShutdown(req)
       context.reply(response)
 
@@ -59,7 +60,8 @@ case class SearchRequest(
     tableInfo: TableInfo,
     projectColumns: Array[String],
     filterExpression: Expression,
-    limit: Long)
+    limit: Long,
+    dataMap: Option[DataMapExprWrapper])
 
 // Search result sent from worker to master
 case class SearchResult(


Mime
View raw message