carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From qiang...@apache.org
Subject [5/5] carbondata git commit: [CARBONDATA-2705][CarbonStore] CarbonStore Java API and Implementation
Date Wed, 11 Jul 2018 08:08:59 GMT
[CARBONDATA-2705][CarbonStore] CarbonStore Java API and Implementation

Support two implementations:
1.LocalCarbonStore for usage in local mode
2.DistributedCarbonStore leveraging multiple server (Master and Workers) via RPC

This closes #2473


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

Branch: refs/heads/carbonstore
Commit: 044a995aa90220715daef83d6628aacea5dd5106
Parents: e4bfb57
Author: Jacky Li <jacky.likun@qq.com>
Authored: Mon Jul 9 12:23:49 2018 +0800
Committer: QiangCai <qiangcai@qq.com>
Committed: Wed Jul 11 16:07:32 2018 +0800

----------------------------------------------------------------------
 .../core/datastore/impl/FileFactory.java        |   5 +
 .../schema/table/TableSchemaBuilder.java        |   3 +-
 .../detailquery/SearchModeTestCase.scala        |   3 +-
 .../carbondata/store/SparkCarbonStore.scala     | 203 --------
 .../apache/carbondata/store/WorkerManager.scala |  75 +++
 .../org/apache/spark/sql/CarbonSession.scala    |  59 ++-
 .../carbondata/store/SparkCarbonStoreTest.scala |  86 ---
 .../processing/loading/DataLoadExecutor.java    |   6 +-
 .../processing/util/CarbonLoaderUtil.java       |   2 +-
 store/core/pom.xml                              |   5 +
 .../carbondata/store/CarbonRowReadSupport.java  |  53 --
 .../apache/carbondata/store/CarbonStore.java    |  68 ---
 .../carbondata/store/LocalCarbonStore.java      | 130 -----
 .../carbondata/store/MetaCachedCarbonStore.java |  59 ---
 .../carbondata/store/api/CarbonStore.java       |  32 ++
 .../store/api/CarbonStoreFactory.java           |  93 ++++
 .../apache/carbondata/store/api/DataStore.java  |  51 ++
 .../apache/carbondata/store/api/MetaStore.java  |  50 ++
 .../apache/carbondata/store/api/SqlStore.java   |  34 ++
 .../carbondata/store/api/conf/StoreConf.java    | 191 +++++++
 .../store/api/descriptor/LoadDescriptor.java    | 114 ++++
 .../store/api/descriptor/SelectDescriptor.java  | 111 ++++
 .../store/api/descriptor/TableDescriptor.java   | 174 +++++++
 .../store/api/descriptor/TableIdentifier.java   |  37 ++
 .../exception/ExecutionTimeoutException.java    |  22 +
 .../store/api/exception/SchedulerException.java |  26 +
 .../store/api/exception/StoreException.java     |  33 ++
 .../apache/carbondata/store/conf/StoreConf.java | 185 -------
 .../exception/ExecutionTimeoutException.java    |  22 -
 .../store/exception/StoreException.java         |  29 --
 .../store/exception/WorkerTooBusyException.java |  26 -
 .../carbondata/store/impl/CarbonStoreBase.java  | 177 +++++++
 .../store/impl/DistributedCarbonStore.java      | 232 +++++++++
 .../store/impl/IndexedRecordReader.java         | 183 +++++++
 .../carbondata/store/impl/LocalCarbonStore.java | 164 ++++++
 .../carbondata/store/impl/MetaProcessor.java    | 170 ++++++
 .../store/impl/SegmentTxnManager.java           | 121 +++++
 .../apache/carbondata/store/impl/Status.java    |  28 +
 .../carbondata/store/impl/master/Master.java    | 161 ++++++
 .../store/impl/master/RegistryServiceImpl.java  |  53 ++
 .../store/impl/master/Schedulable.java          |  76 +++
 .../carbondata/store/impl/master/Scheduler.java | 137 +++++
 .../store/impl/rpc/RegistryService.java         |  32 ++
 .../store/impl/rpc/ServiceFactory.java          |  43 ++
 .../carbondata/store/impl/rpc/StoreService.java |  40 ++
 .../store/impl/rpc/model/BaseResponse.java      |  69 +++
 .../store/impl/rpc/model/LoadDataRequest.java   |  60 +++
 .../store/impl/rpc/model/QueryResponse.java     |  73 +++
 .../impl/rpc/model/RegisterWorkerRequest.java   |  73 +++
 .../impl/rpc/model/RegisterWorkerResponse.java  |  54 ++
 .../carbondata/store/impl/rpc/model/Scan.java   | 108 ++++
 .../store/impl/rpc/model/ShutdownRequest.java   |  53 ++
 .../store/impl/rpc/model/ShutdownResponse.java  |  61 +++
 .../store/impl/worker/RequestHandler.java       | 166 ++++++
 .../store/impl/worker/StoreServiceImpl.java     |  77 +++
 .../carbondata/store/impl/worker/Worker.java    | 166 ++++++
 .../apache/carbondata/store/master/Master.java  | 522 -------------------
 .../carbondata/store/rpc/RegistryService.java   |  32 --
 .../carbondata/store/rpc/ServiceFactory.java    |  43 --
 .../carbondata/store/rpc/StoreService.java      |  40 --
 .../store/rpc/impl/IndexedRecordReader.java     | 183 -------
 .../store/rpc/impl/RegistryServiceImpl.java     |  54 --
 .../store/rpc/impl/RequestHandler.java          | 218 --------
 .../carbondata/store/rpc/impl/Status.java       |  28 -
 .../store/rpc/impl/StoreServiceImpl.java        |  78 ---
 .../store/rpc/model/BaseResponse.java           |  69 ---
 .../store/rpc/model/LoadDataRequest.java        |  60 ---
 .../store/rpc/model/QueryRequest.java           | 108 ----
 .../store/rpc/model/QueryResponse.java          |  73 ---
 .../store/rpc/model/RegisterWorkerRequest.java  |  73 ---
 .../store/rpc/model/RegisterWorkerResponse.java |  54 --
 .../store/rpc/model/ShutdownRequest.java        |  53 --
 .../store/rpc/model/ShutdownResponse.java       |  61 ---
 .../carbondata/store/scheduler/Schedulable.java |  74 ---
 .../carbondata/store/scheduler/Scheduler.java   | 136 -----
 .../apache/carbondata/store/util/StoreUtil.java |   4 +-
 .../apache/carbondata/store/worker/Worker.java  | 166 ------
 .../store/DistributedCarbonStoreTest.java       | 142 +++++
 .../carbondata/store/LocalCarbonStoreTest.java  | 111 +++-
 .../org/apache/carbondata/store/TestUtil.java   | 101 +---
 store/core/src/test/resources/data1.csv         |  11 +
 store/horizon/pom.xml                           |  10 -
 store/horizon/src/main/anltr/Expression.g4      | 163 ------
 store/horizon/src/main/antlr/Expression.g4      | 163 ++++++
 .../apache/carbondata/horizon/antlr/Parser.java |  42 ++
 .../horizon/antlr/gen/ExpressionLexer.java      |   3 -
 .../horizon/antlr/gen/ExpressionParser.java     |   3 -
 .../horizon/rest/client/HorizonClient.java      |  77 +++
 .../rest/client/impl/SimpleHorizonClient.java   |  87 ++++
 .../horizon/rest/controller/Horizon.java        |   6 +-
 .../rest/controller/HorizonController.java      |  76 ++-
 .../rest/model/descriptor/LoadDescriptor.java   |  81 ---
 .../rest/model/descriptor/SelectDescriptor.java |  88 ----
 .../rest/model/descriptor/TableDescriptor.java  |  90 ----
 .../rest/model/validate/RequestValidator.java   |  15 +-
 .../rest/model/view/CreateTableRequest.java     |  13 +-
 .../rest/model/view/DropTableRequest.java       |  45 ++
 .../horizon/rest/model/view/LoadRequest.java    |   8 +-
 .../horizon/rest/model/view/Request.java        |  30 ++
 .../horizon/rest/model/view/Response.java       |  33 ++
 .../horizon/rest/model/view/SelectRequest.java  |  20 +-
 .../horizon/rest/model/view/SelectResponse.java |  18 +-
 .../horizon/rest/service/HorizonService.java    | 162 ------
 .../carbondata/horizon/FilterParseTest.java     |  18 +-
 .../apache/carbondata/horizon/HorizonTest.java  | 165 +++---
 store/horizon/src/test/resources/data1.csv      |  11 -
 .../sdk/file/CarbonWriterBuilder.java           |  11 +-
 .../org/apache/carbondata/sdk/file/Field.java   |  65 +--
 108 files changed, 4624 insertions(+), 3937 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java b/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
index 5c46bcf..bb5d449 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
@@ -220,6 +220,11 @@ public final class FileFactory {
       final FsPermission permission) throws IOException {
     return getCarbonFile(filePath).createNewFile(filePath, fileType, doAs, permission);
   }
+
+  public static boolean deleteFile(String filePath) throws IOException {
+    return deleteFile(filePath, getFileType(filePath));
+  }
+
   public static boolean deleteFile(String filePath, FileType fileType) throws IOException {
     return getCarbonFile(filePath).deleteFile(filePath, fileType);
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
index e4b072b..545e4ba 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchemaBuilder.java
@@ -124,8 +124,9 @@ public class TableSchemaBuilder {
     return schema;
   }
 
-  public void setSortColumns(List<ColumnSchema> sortColumns) {
+  public TableSchemaBuilder setSortColumns(List<ColumnSchema> sortColumns) {
     this.sortColumns = sortColumns;
+    return this;
   }
 
   public ColumnSchema addColumn(StructField field, AtomicInteger valIndex, boolean isSortColumn) {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/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 af9e50f..0b9b6ba 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
@@ -19,7 +19,7 @@ package org.apache.carbondata.spark.testsuite.detailquery
 
 import org.apache.spark.sql.test.util.QueryTest
 import org.apache.spark.sql.{CarbonSession, Row, SaveMode}
-import org.scalatest.BeforeAndAfterAll
+import org.scalatest.{BeforeAndAfterAll, Ignore}
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.util.CarbonProperties
@@ -28,7 +28,6 @@ import org.apache.carbondata.spark.util.DataGenerator
 /**
  * Test Suite for search mode
  */
-
 class SearchModeTestCase extends QueryTest with BeforeAndAfterAll {
 
   val numRows = 500 * 1000

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/integration/spark2/src/main/scala/org/apache/carbondata/store/SparkCarbonStore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/store/SparkCarbonStore.scala b/integration/spark2/src/main/scala/org/apache/carbondata/store/SparkCarbonStore.scala
deleted file mode 100644
index a4124a2..0000000
--- a/integration/spark2/src/main/scala/org/apache/carbondata/store/SparkCarbonStore.scala
+++ /dev/null
@@ -1,203 +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.store
-
-import java.io.IOException
-import java.net.InetAddress
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.{CarbonInputMetrics, SparkConf, SparkEnv}
-import org.apache.spark.sql.CarbonSession._
-import org.apache.spark.sql.SparkSession
-
-import org.apache.carbondata.common.annotations.InterfaceAudience
-import org.apache.carbondata.common.logging.LogServiceFactory
-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.CarbonProjection
-import org.apache.carbondata.spark.rdd.CarbonScanRDD
-import org.apache.carbondata.spark.util.Util
-import org.apache.carbondata.store.conf.StoreConf
-import org.apache.carbondata.store.master.Master
-import org.apache.carbondata.store.worker.Worker
-
-/**
- * A CarbonStore implementation that uses Spark as underlying compute engine
- * with CarbonData query optimization capability
- */
-@InterfaceAudience.Internal
-class SparkCarbonStore extends MetaCachedCarbonStore {
-  private var session: SparkSession = _
-  private var master: Master = _
-  private final val LOG = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  /**
-   * Initialize SparkCarbonStore
-   * @param storeName store name
-   * @param storeLocation location to store data
-   */
-  def this(storeName: String, storeLocation: String) = {
-    this()
-    val sparkConf = new SparkConf(loadDefaults = true)
-    session = SparkSession.builder
-      .config(sparkConf)
-      .appName("SparkCarbonStore-" + storeName)
-      .config("spark.sql.warehouse.dir", storeLocation)
-      .getOrCreateCarbonSession()
-  }
-
-  def this(sparkSession: SparkSession) = {
-    this()
-    session = sparkSession
-  }
-
-  @throws[IOException]
-  override def scan(
-      path: String,
-      projectColumns: Array[String]): java.util.Iterator[CarbonRow] = {
-    require(path != null)
-    require(projectColumns != null)
-    scan(path, projectColumns, null)
-  }
-
-  @throws[IOException]
-  override def scan(
-      path: String,
-      projectColumns: Array[String],
-      filter: Expression): java.util.Iterator[CarbonRow] = {
-    require(path != null)
-    require(projectColumns != null)
-    val table = getTable(path)
-    val rdd = new CarbonScanRDD[CarbonRow](
-      spark = session,
-      columnProjection = new CarbonProjection(projectColumns),
-      filterExpression = filter,
-      identifier = table.getAbsoluteTableIdentifier,
-      serializedTableInfo = table.getTableInfo.serialize,
-      tableInfo = table.getTableInfo,
-      inputMetricsStats = new CarbonInputMetrics,
-      partitionNames = null,
-      dataTypeConverterClz = null,
-      readSupportClz = classOf[CarbonRowReadSupport])
-    rdd.collect
-      .iterator
-      .asJava
-  }
-
-  @throws[IOException]
-  override def sql(sqlString: String): java.util.Iterator[CarbonRow] = {
-    val df = session.sql(sqlString)
-    df.rdd
-      .map(row => new CarbonRow(row.toSeq.toArray.asInstanceOf[Array[Object]]))
-      .collect()
-      .iterator
-      .asJava
-  }
-
-  def startSearchMode(): Unit = {
-    LOG.info("Starting search mode master")
-    val conf = new StoreConf()
-    conf.conf(StoreConf.MASTER_HOST, InetAddress.getLocalHost.getHostAddress)
-    conf.conf(StoreConf.MASTER_PORT, CarbonProperties.getSearchMasterPort)
-    conf.conf(StoreConf.STORE_LOCATION, CarbonProperties.getStorePath)
-    master = Master.getInstance(conf)
-    master.startService()
-    startAllWorkers()
-  }
-
-  def stopSearchMode(): Unit = {
-    if (master != null) {
-      LOG.info("Shutting down all workers...")
-      try {
-        master.stopAllWorkers()
-        LOG.info("All workers are shut down")
-      } catch {
-        case e: Exception =>
-          LOG.error(s"failed to shutdown worker: ${ e.toString }")
-      }
-      LOG.info("Stopping master...")
-      master.stopService()
-      LOG.info("Master stopped")
-      master = null
-    }
-  }
-
-  /** search mode */
-  def search(
-      table: CarbonTable,
-      projectColumns: Array[String],
-      filter: Expression,
-      globalLimit: Long,
-      localLimit: Long): java.util.Iterator[CarbonRow] = {
-    if (master == null) {
-      throw new IllegalStateException("search mode is not started")
-    }
-    master.search(table, projectColumns, filter, globalLimit, localLimit)
-      .iterator
-      .asJava
-  }
-
-  private def startAllWorkers(): Array[Int] = {
-    // TODO: how to ensure task is sent to every executor?
-    val numExecutors = session.sparkContext.getExecutorMemoryStatus.keySet.size
-    val masterIp = InetAddress.getLocalHost.getHostAddress
-    val rows = session.sparkContext.parallelize(1 to numExecutors * 10, numExecutors)
-      .mapPartitions { f =>
-        // start worker
-        val conf = new StoreConf()
-        conf.conf(StoreConf.WORKER_HOST, InetAddress.getLocalHost.getHostAddress)
-        conf.conf(StoreConf.WORKER_PORT, CarbonProperties.getSearchWorkerPort)
-        conf.conf(StoreConf.WORKER_CORE_NUM, 2)
-        conf.conf(StoreConf.STORE_LOCATION, CarbonProperties.getStorePath)
-        conf.conf(StoreConf.MASTER_HOST, masterIp)
-        conf.conf(StoreConf.MASTER_PORT, CarbonProperties.getSearchMasterPort)
-
-        var storeLocation: String = null
-        val carbonUseLocalDir = CarbonProperties.getInstance()
-          .getProperty("carbon.use.local.dir", "false")
-        if (carbonUseLocalDir.equalsIgnoreCase("true")) {
-
-          val storeLocations = Util.getConfiguredLocalDirs(SparkEnv.get.conf)
-          if (null != storeLocations && storeLocations.nonEmpty) {
-            storeLocation = storeLocations.mkString(",")
-          }
-          if (storeLocation == null) {
-            storeLocation = System.getProperty("java.io.tmpdir")
-          }
-        } else {
-          storeLocation = System.getProperty("java.io.tmpdir")
-        }
-        conf.conf(StoreConf.STORE_TEMP_LOCATION, storeLocation)
-
-        val worker = new Worker(conf)
-        worker.start()
-        new Iterator[Int] {
-          override def hasNext: Boolean = false
-
-          override def next(): Int = 1
-        }
-      }.collect()
-    LOG.info(s"Tried to start $numExecutors workers, ${master.getWorkers.size} " +
-             s"workers are started successfully")
-    rows
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/integration/spark2/src/main/scala/org/apache/carbondata/store/WorkerManager.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/store/WorkerManager.scala b/integration/spark2/src/main/scala/org/apache/carbondata/store/WorkerManager.scala
new file mode 100644
index 0000000..7fff2e5
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/store/WorkerManager.scala
@@ -0,0 +1,75 @@
+/*
+ * 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.store
+
+import java.net.InetAddress
+
+import org.apache.spark.SparkEnv
+import org.apache.spark.sql.SparkSession
+
+import org.apache.carbondata.common.annotations.InterfaceAudience
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.spark.util.Util
+import org.apache.carbondata.store.api.CarbonStoreFactory
+import org.apache.carbondata.store.api.conf.StoreConf
+import org.apache.carbondata.store.impl.worker.Worker
+
+/**
+ * A CarbonStore implementation that uses Spark as underlying compute engine
+ * with CarbonData query optimization capability
+ */
+@InterfaceAudience.Internal
+object WorkerManager {
+  private final val LOG = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+
+  // TODO: how to ensure task is sent to every executor?
+  def startAllWorker(session: SparkSession, storeConf: StoreConf): Unit = {
+    val numExecutors = session.sparkContext.getExecutorMemoryStatus.keySet.size
+    LOG.info("Starting search mode master")
+    val rows = session.sparkContext.parallelize(1 to numExecutors * 10, numExecutors)
+      .mapPartitions { f =>
+        // start worker
+        var storeLocation: String = null
+        val carbonUseLocalDir = CarbonProperties.getInstance()
+          .getProperty("carbon.use.local.dir", "false")
+        if (carbonUseLocalDir.equalsIgnoreCase("true")) {
+          val storeLocations = Util.getConfiguredLocalDirs(SparkEnv.get.conf)
+          if (null != storeLocations && storeLocations.nonEmpty) {
+            storeLocation = storeLocations.mkString(",")
+          }
+          if (storeLocation == null) {
+            storeLocation = System.getProperty("java.io.tmpdir")
+          }
+        } else {
+          storeLocation = System.getProperty("java.io.tmpdir")
+        }
+        storeConf.conf(StoreConf.STORE_TEMP_LOCATION, storeLocation)
+
+        val worker = new Worker(storeConf)
+        worker.start()
+        new Iterator[Int] {
+          override def hasNext: Boolean = false
+
+          override def next(): Int = 1
+        }
+      }.collect()
+    LOG.info(s"Tried to start $numExecutors workers, $rows workers started successfully")
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/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 ee19503..cd939dc 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
@@ -17,6 +17,7 @@
 package org.apache.spark.sql
 
 import java.io.File
+import java.net.InetAddress
 import java.util.concurrent.atomic.AtomicLong
 
 import scala.collection.JavaConverters._
@@ -41,7 +42,10 @@ import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.util.{CarbonProperties, CarbonSessionInfo, ThreadLocalSessionInfo}
 import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
-import org.apache.carbondata.store.SparkCarbonStore
+import org.apache.carbondata.store.WorkerManager
+import org.apache.carbondata.store.api.{CarbonStore, CarbonStoreFactory}
+import org.apache.carbondata.store.api.conf.StoreConf
+import org.apache.carbondata.store.api.descriptor.{SelectDescriptor, TableIdentifier => CTableIdentifier}
 import org.apache.carbondata.streaming.CarbonStreamingQueryListener
 
 /**
@@ -120,7 +124,7 @@ class CarbonSession(@transient val sc: SparkContext,
     message(0).getString(0).contains(dataMapName)
   }
 
-  def isSearchModeEnabled: Boolean = carbonStore != null
+  def isSearchModeEnabled: Boolean = store != null
 
   /**
    * Run SparkSQL directly
@@ -192,24 +196,34 @@ class CarbonSession(@transient val sc: SparkContext,
     }
   }
 
-  @transient private var carbonStore: SparkCarbonStore = _
+  // variable that used in search mode
+  @transient private var store: CarbonStore = _
 
   def startSearchMode(): Unit = {
-    CarbonProperties.enableSearchMode(true)
-    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "false")
-    if (carbonStore == null) {
-      carbonStore = new SparkCarbonStore(this)
-      carbonStore.startSearchMode()
+    if (store == null) {
+      val storeConf = new StoreConf()
+      storeConf.conf(StoreConf.STORE_LOCATION, CarbonProperties.getStorePath)
+      storeConf.conf(StoreConf.MASTER_HOST, InetAddress.getLocalHost.getHostAddress)
+      storeConf.conf(StoreConf.MASTER_PORT, CarbonProperties.getSearchMasterPort)
+      storeConf.conf(StoreConf.WORKER_HOST, InetAddress.getLocalHost.getHostAddress)
+      storeConf.conf(StoreConf.WORKER_PORT, CarbonProperties.getSearchWorkerPort)
+      storeConf.conf(StoreConf.WORKER_CORE_NUM, 2)
+
+      store = CarbonStoreFactory.getDistributedStore("GlobalStore", storeConf)
+      CarbonProperties.enableSearchMode(true)
+      CarbonProperties.getInstance()
+        .addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "false")
+      WorkerManager.startAllWorker(this, storeConf)
     }
   }
 
   def stopSearchMode(): Unit = {
-    CarbonProperties.enableSearchMode(false)
-    CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "true")
-    if (carbonStore != null) {
+    if (store != null) {
+      CarbonProperties.enableSearchMode(false)
+      CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_VECTOR_READER, "true")
       try {
-        carbonStore.stopSearchMode()
-        carbonStore = null
+        CarbonStoreFactory.removeDistributedStore("GlobalStore")
+        store = null
       } catch {
         case e: RuntimeException =>
           LogServiceFactory.getLogService(this.getClass.getCanonicalName)
@@ -226,16 +240,21 @@ class CarbonSession(@transient val sc: SparkContext,
       relation: LogicalRelation,
       maxRows: Option[Long] = None,
       localMaxRows: Option[Long] = None): DataFrame = {
-    val rows = carbonStore.search(
-        relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable,
-        columns.map(_.name).toArray,
-        if (expr != null) CarbonFilters.transformExpression(expr) else null,
-        maxRows.getOrElse(Long.MaxValue),
-        localMaxRows.getOrElse(Long.MaxValue))
+    val table = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable
+    val select = new SelectDescriptor(
+      new CTableIdentifier(table.getTableName, table.getDatabaseName),
+      columns.map(_.name).toArray,
+      if (expr != null) CarbonFilters.transformExpression(expr) else null,
+      localMaxRows.getOrElse(Long.MaxValue)
+    )
+    val rows = store.select(select).iterator()
     val output = new java.util.ArrayList[Row]()
-    while (rows.hasNext) {
+    val maxRowCount = maxRows.getOrElse(Long.MaxValue)
+    var rowCount = 0
+    while (rows.hasNext && rowCount < maxRowCount) {
       val row = rows.next()
       output.add(Row.fromSeq(row.getData))
+      rowCount = rowCount + 1
     }
     createDataFrame(output, logicalPlan.schema)
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/integration/spark2/src/test/scala/org/apache/carbondata/store/SparkCarbonStoreTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/store/SparkCarbonStoreTest.scala b/integration/spark2/src/test/scala/org/apache/carbondata/store/SparkCarbonStoreTest.scala
deleted file mode 100644
index d389670..0000000
--- a/integration/spark2/src/test/scala/org/apache/carbondata/store/SparkCarbonStoreTest.scala
+++ /dev/null
@@ -1,86 +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.store
-
-import org.apache.spark.sql.{CarbonEnv, Row}
-import org.apache.spark.sql.test.util.QueryTest
-import org.scalatest.BeforeAndAfterAll
-
-import org.apache.carbondata.core.metadata.datatype.DataTypes
-import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression
-import org.apache.carbondata.core.scan.expression.{ColumnExpression, LiteralExpression}
-
-class SparkCarbonStoreTest extends QueryTest with BeforeAndAfterAll {
-
-  private var store: CarbonStore = _
-
-  override def beforeAll {
-    sql("DROP TABLE IF EXISTS t1")
-    sql("CREATE TABLE t1 (" +
-        "empno int, empname String, designation String, doj Timestamp, " +
-        "workgroupcategory int, workgroupcategoryname String, deptno int, deptname String," +
-        "projectcode int, projectjoindate Timestamp, projectenddate Timestamp," +
-        "attendance int,utilization int,salary int)" +
-        "STORED BY 'org.apache.carbondata.format'")
-    sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/data.csv' INTO TABLE t1 OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '\"')""")
-
-    store = new SparkCarbonStore("test", storeLocation)
-  }
-
-  test("test CarbonStore.get, compare projection result") {
-    val tablePath = CarbonEnv.getCarbonTable(None, "t1")(sqlContext.sparkSession).getTablePath
-    val rows = store.scan(s"$tablePath", Seq("empno", "empname").toArray)
-    val sparkResult: Array[Row] = sql("select empno, empname from t1").collect()
-    sparkResult.zipWithIndex.foreach { case (r: Row, i: Int) =>
-      val carbonRow = rows.next()
-      assertResult(r.get(0))(carbonRow.getData()(0))
-      assertResult(r.get(1))(carbonRow.getData()(1))
-    }
-    assert(!rows.hasNext)
-  }
-
-  test("test CarbonStore.get, compare projection and filter result") {
-    val tablePath = CarbonEnv.getCarbonTable(None, "t1")(sqlContext.sparkSession).getTablePath
-    val filter = new EqualToExpression(
-      new ColumnExpression("empno", DataTypes.INT),
-      new LiteralExpression(10, DataTypes.INT))
-    val rows = store.scan(s"$tablePath", Seq("empno", "empname").toArray, filter)
-    val sparkResult: Array[Row] = sql("select empno, empname from t1 where empno = 10").collect()
-    sparkResult.zipWithIndex.foreach { case (r: Row, i: Int) =>
-      val carbonRow = rows.next()
-      assertResult(r.get(0))(carbonRow.getData()(0))
-      assertResult(r.get(1))(carbonRow.getData()(1))
-    }
-    assert(!rows.hasNext)
-  }
-
-  test("test CarbonStore.sql") {
-    val rows = store.sql("select empno, empname from t1 where empno = 10")
-    val sparkResult: Array[Row] = sql("select empno, empname from t1 where empno = 10").collect()
-    sparkResult.zipWithIndex.foreach { case (r: Row, i: Int) =>
-      val carbonRow = rows.next()
-      assertResult(r.get(0))(carbonRow.getData()(0))
-      assertResult(r.get(1))(carbonRow.getData()(1))
-    }
-    assert(!rows.hasNext)
-  }
-
-  override def afterAll {
-    sql("DROP TABLE IF EXISTS t1")
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadExecutor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadExecutor.java b/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadExecutor.java
index fc5c41f..65d72cc 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadExecutor.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/DataLoadExecutor.java
@@ -39,11 +39,11 @@ public class DataLoadExecutor {
 
   private boolean isClosed;
 
-  public void execute(CarbonLoadModel loadModel, String[] storeLocation,
-      CarbonIterator<Object[]>[] inputIterators) throws Exception {
+  public void execute(CarbonLoadModel loadModel, String[] tempLocation,
+      CarbonIterator<Object[]>[] inputIterators) {
     try {
       loadProcessorStep =
-          new DataLoadProcessBuilder().build(loadModel, storeLocation, inputIterators);
+          new DataLoadProcessBuilder().build(loadModel, tempLocation, inputIterators);
       // 1. initialize
       loadProcessorStep.initialize();
       LOGGER.info("Data Loading is started for table " + loadModel.getTableName());

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
index 29ce2f4..0460d16 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
@@ -578,7 +578,7 @@ public final class CarbonLoaderUtil {
    *                       based on block location information
    * @param blockAssignmentStrategy strategy used to assign blocks
    * @param expectedMinSizePerNode expected minimum size per node
-   * @return a map that maps node to blocks
+   * @return a map that maps node (hostname) to blocks
    */
   public static Map<String, List<Distributable>> nodeBlockMapping(
       List<Distributable> blockInfos, int noOfNodesInput, List<String> activeNodes,

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/pom.xml
----------------------------------------------------------------------
diff --git a/store/core/pom.xml b/store/core/pom.xml
index 6b2703e..44d5ab1 100644
--- a/store/core/pom.xml
+++ b/store/core/pom.xml
@@ -26,6 +26,11 @@
       <version>${project.version}</version>
     </dependency>
     <dependency>
+      <groupId>org.antlr</groupId>
+      <artifactId>antlr4-runtime</artifactId>
+      <version>4.7.1</version>
+    </dependency>
+    <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/CarbonRowReadSupport.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/CarbonRowReadSupport.java b/store/core/src/main/java/org/apache/carbondata/store/CarbonRowReadSupport.java
deleted file mode 100644
index bafbb9f..0000000
--- a/store/core/src/main/java/org/apache/carbondata/store/CarbonRowReadSupport.java
+++ /dev/null
@@ -1,53 +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.store;
-
-import java.io.IOException;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.core.datastore.row.CarbonRow;
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
-import org.apache.carbondata.hadoop.readsupport.CarbonReadSupport;
-import org.apache.carbondata.hadoop.readsupport.impl.DictionaryDecodeReadSupport;
-
-/**
- * ReadSupport that convert row object to CarbonRow
- */
-@InterfaceAudience.Internal
-public class CarbonRowReadSupport implements CarbonReadSupport<CarbonRow> {
-  private CarbonReadSupport<Object[]> delegate;
-
-  public CarbonRowReadSupport() {
-    this.delegate = new DictionaryDecodeReadSupport<>();
-  }
-
-  @Override public void initialize(CarbonColumn[] carbonColumns, CarbonTable carbonTable)
-      throws IOException {
-    delegate.initialize(carbonColumns, carbonTable);
-  }
-
-  @Override public CarbonRow readRow(Object[] data) {
-    Object[] converted = delegate.readRow(data);
-    return new CarbonRow(converted);
-  }
-
-  @Override public void close() {
-    delegate.close();
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/CarbonStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/CarbonStore.java b/store/core/src/main/java/org/apache/carbondata/store/CarbonStore.java
deleted file mode 100644
index c6b2fb8..0000000
--- a/store/core/src/main/java/org/apache/carbondata/store/CarbonStore.java
+++ /dev/null
@@ -1,68 +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.store;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Iterator;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.common.annotations.InterfaceStability;
-import org.apache.carbondata.core.datastore.row.CarbonRow;
-import org.apache.carbondata.core.scan.expression.Expression;
-
-/**
- * User can use {@link CarbonStore} to query data
- */
-@InterfaceAudience.User
-@InterfaceStability.Unstable
-public interface CarbonStore extends Closeable {
-
-  /**
-   * Scan query on the data in the table path
-   * @param path table path
-   * @param projectColumns column names to read
-   * @return rows
-   * @throws IOException if unable to read files in table path
-   */
-  Iterator<CarbonRow> scan(
-      String path,
-      String[] projectColumns) throws IOException;
-
-  /**
-   * Scan query with filter, on the data in the table path
-   * @param path table path
-   * @param projectColumns column names to read
-   * @param filter filter condition, can be null
-   * @return rows that satisfy filter condition
-   * @throws IOException if unable to read files in table path
-   */
-  Iterator<CarbonRow> scan(
-      String path,
-      String[] projectColumns,
-      Expression filter) throws IOException;
-
-  /**
-   * SQL query, table should be created before calling this function
-   * @param sqlString SQL statement
-   * @return rows
-   * @throws IOException if unable to read files in table path
-   */
-  Iterator<CarbonRow> sql(String sqlString) throws IOException;
-
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/LocalCarbonStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/LocalCarbonStore.java b/store/core/src/main/java/org/apache/carbondata/store/LocalCarbonStore.java
deleted file mode 100644
index daa1447..0000000
--- a/store/core/src/main/java/org/apache/carbondata/store/LocalCarbonStore.java
+++ /dev/null
@@ -1,130 +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.store;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-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.datastore.row.CarbonRow;
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
-import org.apache.carbondata.core.scan.expression.Expression;
-import org.apache.carbondata.hadoop.CarbonProjection;
-import org.apache.carbondata.hadoop.api.CarbonInputFormat;
-import org.apache.carbondata.hadoop.api.CarbonTableInputFormat;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobID;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.hadoop.mapreduce.task.JobContextImpl;
-import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
-
-/**
- * A CarbonStore implementation that works locally, without other compute framework dependency.
- * It can be used to read data in local disk.
- *
- * Note that this class is experimental, it is not intended to be used in production.
- */
-@InterfaceAudience.Internal
-class LocalCarbonStore extends MetaCachedCarbonStore {
-
-  private static final LogService LOGGER =
-      LogServiceFactory.getLogService(LocalCarbonStore.class.getName());
-
-  @Override
-  public Iterator<CarbonRow> scan(String path, String[] projectColumns) throws IOException {
-    return scan(path, projectColumns, null);
-  }
-
-  @Override public Iterator<CarbonRow> scan(String path, String[] projectColumns, Expression filter)
-      throws IOException {
-    Objects.requireNonNull(path);
-    Objects.requireNonNull(projectColumns);
-
-    CarbonTable table = getTable(path);
-    if (table.isStreamingSink() || table.isHivePartitionTable()) {
-      throw new UnsupportedOperationException("streaming and partition table is not supported");
-    }
-    // TODO: use InputFormat to prune data and read data
-
-    final CarbonTableInputFormat format = new CarbonTableInputFormat();
-    final Job job = new Job(new Configuration());
-    CarbonInputFormat.setTableInfo(job.getConfiguration(), table.getTableInfo());
-    CarbonInputFormat.setTablePath(job.getConfiguration(), table.getTablePath());
-    CarbonInputFormat.setTableName(job.getConfiguration(), table.getTableName());
-    CarbonInputFormat.setDatabaseName(job.getConfiguration(), table.getDatabaseName());
-    CarbonInputFormat.setCarbonReadSupport(job.getConfiguration(), CarbonRowReadSupport.class);
-    CarbonInputFormat
-        .setColumnProjection(job.getConfiguration(), new CarbonProjection(projectColumns));
-    if (filter != null) {
-      CarbonInputFormat.setFilterPredicates(job.getConfiguration(), filter);
-    }
-
-    final List<InputSplit> splits =
-        format.getSplits(new JobContextImpl(job.getConfiguration(), new JobID()));
-
-    List<RecordReader<Void, Object>> readers = new ArrayList<>(splits.size());
-
-    List<CarbonRow> rows = new ArrayList<>();
-
-    try {
-      for (InputSplit split : splits) {
-        TaskAttemptContextImpl attempt =
-            new TaskAttemptContextImpl(job.getConfiguration(), new TaskAttemptID());
-        RecordReader reader = format.createRecordReader(split, attempt);
-        reader.initialize(split, attempt);
-        readers.add(reader);
-      }
-
-      for (RecordReader<Void, Object> reader : readers) {
-        while (reader.nextKeyValue()) {
-          rows.add((CarbonRow) reader.getCurrentValue());
-        }
-        try {
-          reader.close();
-        } catch (IOException e) {
-          LOGGER.error(e);
-        }
-      }
-    } catch (InterruptedException e) {
-      throw new IOException(e);
-    } finally {
-      for (RecordReader<Void, Object> reader : readers) {
-        try {
-          reader.close();
-        } catch (IOException e) {
-          LOGGER.error(e);
-        }
-      }
-    }
-    return rows.iterator();
-  }
-
-  @Override
-  public Iterator<CarbonRow> sql(String sqlString) throws IOException {
-    throw new UnsupportedOperationException();
-  }
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java b/store/core/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java
deleted file mode 100644
index e43f750..0000000
--- a/store/core/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java
+++ /dev/null
@@ -1,59 +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.store;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.core.metadata.converter.SchemaConverter;
-import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl;
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
-import org.apache.carbondata.core.metadata.schema.table.TableInfo;
-import org.apache.carbondata.core.util.CarbonUtil;
-import org.apache.carbondata.core.util.path.CarbonTablePath;
-
-/**
- * A CarbonStore base class that caches CarbonTable object
- */
-@InterfaceAudience.Internal
-abstract class MetaCachedCarbonStore implements CarbonStore {
-
-  // mapping of table path to CarbonTable object
-  private Map<String, CarbonTable> cache = new HashMap<>();
-
-  CarbonTable getTable(String path) throws IOException {
-    if (cache.containsKey(path)) {
-      return cache.get(path);
-    }
-    org.apache.carbondata.format.TableInfo tableInfo = CarbonUtil
-        .readSchemaFile(CarbonTablePath.getSchemaFilePath(path));
-    SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
-    TableInfo tableInfo1 = schemaConverter.fromExternalToWrapperTableInfo(tableInfo, "", "", "");
-    tableInfo1.setTablePath(path);
-    CarbonTable table = CarbonTable.buildFromTableInfo(tableInfo1);
-    cache.put(path, table);
-    return table;
-  }
-
-  @Override
-  public void close() throws IOException {
-    cache.clear();
-  }
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStore.java b/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStore.java
new file mode 100644
index 0000000..3525389
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStore.java
@@ -0,0 +1,32 @@
+/*
+ * 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.store.api;
+
+import java.io.Closeable;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.annotations.InterfaceStability;
+
+/**
+ * Public Interface of CarbonStore
+ */
+@InterfaceAudience.User
+@InterfaceStability.Unstable
+public interface CarbonStore extends MetaStore, DataStore, Closeable {
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStoreFactory.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStoreFactory.java b/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStoreFactory.java
new file mode 100644
index 0000000..76ef450
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/CarbonStoreFactory.java
@@ -0,0 +1,93 @@
+/*
+ * 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.store.api;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.store.api.conf.StoreConf;
+import org.apache.carbondata.store.api.exception.StoreException;
+
+public class CarbonStoreFactory {
+  private static Map<String, CarbonStore> distributedStores = new ConcurrentHashMap<>();
+  private static Map<String, CarbonStore> localStores = new ConcurrentHashMap<>();
+
+  private CarbonStoreFactory() {
+  }
+
+  public static CarbonStore getDistributedStore(String storeName, StoreConf storeConf)
+      throws StoreException {
+    if (distributedStores.containsKey(storeName)) {
+      return distributedStores.get(storeName);
+    }
+
+    // create a new instance
+    try {
+      String className = "org.apache.carbondata.store.impl.DistributedCarbonStore";
+      CarbonStore store = createCarbonStore(storeConf, className);
+      distributedStores.put(storeName, store);
+      return store;
+    } catch (ClassNotFoundException | IllegalAccessException | InvocationTargetException |
+        InstantiationException e) {
+      throw new StoreException(e);
+    }
+  }
+
+  public static void removeDistributedStore(String storeName) throws IOException {
+    if (distributedStores.containsKey(storeName)) {
+      distributedStores.get(storeName).close();
+      distributedStores.remove(storeName);
+    }
+  }
+
+  public static CarbonStore getLocalStore(String storeName, StoreConf storeConf)
+      throws StoreException {
+    if (localStores.containsKey(storeName)) {
+      return localStores.get(storeName);
+    }
+
+    // create a new instance
+    try {
+      String className = "org.apache.carbondata.store.impl.LocalCarbonStore";
+      CarbonStore store = createCarbonStore(storeConf, className);
+      localStores.put(storeName, store);
+      return store;
+    } catch (ClassNotFoundException | IllegalAccessException | InvocationTargetException |
+        InstantiationException e) {
+      throw new StoreException(e);
+    }
+  }
+
+  public static void removeLocalStore(String storeName) throws IOException {
+    if (localStores.containsKey(storeName)) {
+      localStores.get(storeName).close();
+      localStores.remove(storeName);
+    }
+  }
+
+  private static CarbonStore createCarbonStore(StoreConf storeConf, String className)
+      throws ClassNotFoundException, InstantiationException, IllegalAccessException,
+      InvocationTargetException {
+    Constructor[] constructor = Class.forName(className).getDeclaredConstructors();
+    constructor[0].setAccessible(true);
+    return (CarbonStore) constructor[0].newInstance(storeConf);
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/DataStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/DataStore.java b/store/core/src/main/java/org/apache/carbondata/store/api/DataStore.java
new file mode 100644
index 0000000..d35c133
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/DataStore.java
@@ -0,0 +1,51 @@
+/*
+ * 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.store.api;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.annotations.InterfaceStability;
+import org.apache.carbondata.core.datastore.row.CarbonRow;
+import org.apache.carbondata.store.api.descriptor.LoadDescriptor;
+import org.apache.carbondata.store.api.descriptor.SelectDescriptor;
+import org.apache.carbondata.store.api.exception.StoreException;
+
+/**
+ * Public interface to write and read data in CarbonStore
+ */
+@InterfaceAudience.User
+@InterfaceStability.Unstable
+public interface DataStore {
+
+  /**
+   * Load data into a Table
+   * @param load descriptor for load operation
+   * @throws IOException if network or disk IO error occurs
+   */
+  void loadData(LoadDescriptor load) throws IOException, StoreException;
+
+  /**
+   * Scan a Table and return matched rows
+   * @param select descriptor for scan operation, including required column, filter, etc
+   * @return matched rows
+   * @throws IOException if network or disk IO error occurs
+   */
+  List<CarbonRow> select(SelectDescriptor select) throws IOException, StoreException;
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/MetaStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/MetaStore.java b/store/core/src/main/java/org/apache/carbondata/store/api/MetaStore.java
new file mode 100644
index 0000000..dea6873
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/MetaStore.java
@@ -0,0 +1,50 @@
+/*
+ * 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.store.api;
+
+import java.io.IOException;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.annotations.InterfaceStability;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.store.api.descriptor.TableDescriptor;
+import org.apache.carbondata.store.api.descriptor.TableIdentifier;
+import org.apache.carbondata.store.api.exception.StoreException;
+
+/**
+ * Public interface to manage table in CarbonStore
+ */
+@InterfaceAudience.User
+@InterfaceStability.Unstable
+public interface MetaStore {
+  /**
+   * Create a Table
+   * @param table descriptor for create table operation
+   * @throws IOException if network or disk IO error occurs
+   */
+  void createTable(TableDescriptor table) throws IOException, StoreException;
+
+  /**
+   * Drop a Table, and remove all data in it
+   * @param table table identifier
+   * @throws IOException if network or disk IO error occurs
+   */
+  void dropTable(TableIdentifier table) throws IOException;
+
+  CarbonTable getTable(TableIdentifier table) throws IOException;
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/SqlStore.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/SqlStore.java b/store/core/src/main/java/org/apache/carbondata/store/api/SqlStore.java
new file mode 100644
index 0000000..3f52eed
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/SqlStore.java
@@ -0,0 +1,34 @@
+/*
+ * 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.store.api;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.carbondata.core.datastore.row.CarbonRow;
+
+public interface SqlStore {
+
+  /**
+   * Executor a SQL statement
+   * @param sqlString SQL statement
+   * @return matched rows
+   * @throws IOException if network or disk IO error occurs
+   */
+  List<CarbonRow> sql(String sqlString) throws IOException;
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/conf/StoreConf.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/conf/StoreConf.java b/store/core/src/main/java/org/apache/carbondata/store/api/conf/StoreConf.java
new file mode 100644
index 0000000..98a670a
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/conf/StoreConf.java
@@ -0,0 +1,191 @@
+/*
+ * 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.store.api.conf;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.store.util.StoreUtil;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+public class StoreConf implements Serializable, Writable {
+
+  public static final String SELECT_PROJECTION = "carbon.select.projection";
+  public static final String SELECT_FILTER = "carbon.select.filter";
+  public static final String SELECT_LIMIT = "carbon.select.limit";
+
+  public static final String SELECT_ID = "carbon.select.id";
+
+  public static final String WORKER_HOST = "carbon.worker.host";
+  public static final String WORKER_PORT = "carbon.worker.port";
+  public static final String WORKER_CORE_NUM = "carbon.worker.core.num";
+  public static final String MASTER_HOST = "carbon.master.host";
+  public static final String MASTER_PORT = "carbon.master.port";
+
+  public static final String STORE_TEMP_LOCATION = "carbon.store.temp.location";
+  public static final String STORE_LOCATION = "carbon.store.location";
+  public static final String STORE_NAME = "carbon.store.name";
+
+  private Map<String, String> conf = new HashMap<>();
+
+  public StoreConf() {
+  }
+
+  public StoreConf(String storeName, String storeLocation) {
+    conf.put(STORE_NAME, storeName);
+    conf.put(STORE_LOCATION, storeLocation);
+  }
+
+  public StoreConf(String confFilePath) {
+    load(confFilePath);
+  }
+
+  public StoreConf conf(String key, String value) {
+    conf.put(key, value);
+    return this;
+  }
+
+  public StoreConf conf(String key, int value) {
+    conf.put(key, "" + value);
+    return this;
+  }
+
+  public void load(String filePath) {
+    StoreUtil.loadProperties(filePath, this);
+  }
+
+  public void conf(StoreConf conf) {
+    this.conf.putAll(conf.conf);
+  }
+
+  public Object conf(String key) {
+    return conf.get(key);
+  }
+
+  public String[] projection() {
+    return stringArrayValue(SELECT_PROJECTION);
+  }
+
+  public String filter() {
+    return stringValue(SELECT_FILTER);
+  }
+
+  public int limit() {
+    return intValue(SELECT_LIMIT);
+  }
+
+  public String masterHost() {
+    return stringValue(MASTER_HOST);
+  }
+
+  public int masterPort() {
+    return intValue(MASTER_PORT);
+  }
+
+  public String workerHost() {
+    return stringValue(WORKER_HOST);
+  }
+
+  public int workerPort() {
+    return intValue(WORKER_PORT);
+  }
+
+  public int workerCoreNum() {
+    return intValue(WORKER_CORE_NUM);
+  }
+
+  public String storeLocation() {
+    return stringValue(STORE_LOCATION);
+  }
+
+  public String[] storeTempLocation() {
+    return stringArrayValue(STORE_TEMP_LOCATION);
+  }
+
+  public String selectId() {
+    return stringValue(SELECT_ID);
+  }
+
+  public Configuration newHadoopConf() {
+    Configuration hadoopConf = FileFactory.getConfiguration();
+    for (Map.Entry<String, String> entry : conf.entrySet()) {
+      String key = entry.getKey();
+      String value = entry.getValue();
+      if (key != null && value != null && key.startsWith("carbon.hadoop.")) {
+        hadoopConf.set(key.substring("carbon.hadoop.".length()), value);
+      }
+    }
+    return hadoopConf;
+  }
+
+  private String stringValue(String key) {
+    Object obj = conf.get(key);
+    if (obj == null) {
+      return null;
+    }
+    return obj.toString();
+  }
+
+  private int intValue(String key) {
+    String value = conf.get(key);
+    if (value == null) {
+      return -1;
+    }
+    return Integer.parseInt(value);
+  }
+
+  private String[] stringArrayValue(String key) {
+    String value = conf.get(key);
+    if (value == null) {
+      return null;
+    }
+    return value.split(",", -1);
+  }
+
+  @Override public void write(DataOutput out) throws IOException {
+    Set<Map.Entry<String, String>> entries = conf.entrySet();
+    WritableUtils.writeVInt(out, conf.size());
+    for (Map.Entry<String, String> entry : entries) {
+      WritableUtils.writeString(out, entry.getKey());
+      WritableUtils.writeString(out, entry.getValue());
+    }
+  }
+
+  @Override public void readFields(DataInput in) throws IOException {
+    if (conf == null) {
+      conf = new HashMap<>();
+    }
+
+    int size = WritableUtils.readVInt(in);
+    String key, value;
+    for (int i = 0; i < size; i++) {
+      key = WritableUtils.readString(in);
+      value = WritableUtils.readString(in);
+      conf.put(key, value);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/LoadDescriptor.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/LoadDescriptor.java b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/LoadDescriptor.java
new file mode 100644
index 0000000..c3a4ff7
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/LoadDescriptor.java
@@ -0,0 +1,114 @@
+/*
+ * 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.store.api.descriptor;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+public class LoadDescriptor {
+
+  private TableIdentifier table;
+  private String inputPath;
+  private Map<String, String> options;
+  private boolean isOverwrite;
+
+  private LoadDescriptor() {
+  }
+
+  public LoadDescriptor(TableIdentifier table, String inputPath,
+      Map<String, String> options, boolean isOverwrite) {
+    Objects.requireNonNull(table);
+    Objects.requireNonNull(inputPath);
+    this.table = table;
+    this.inputPath = inputPath;
+    this.options = options;
+    this.isOverwrite = isOverwrite;
+  }
+
+  public TableIdentifier getTable() {
+    return table;
+  }
+
+  public void setTable(TableIdentifier table) {
+    this.table = table;
+  }
+
+  public String getInputPath() {
+    return inputPath;
+  }
+
+  public void setInputPath(String inputPath) {
+    this.inputPath = inputPath;
+  }
+
+  public Map<String, String> getOptions() {
+    return options;
+  }
+
+  public void setOptions(Map<String, String> options) {
+    this.options = options;
+  }
+
+  public boolean isOverwrite() {
+    return isOverwrite;
+  }
+
+  public void setOverwrite(boolean overwrite) {
+    isOverwrite = overwrite;
+  }
+
+  public static class Builder {
+    private LoadDescriptor load;
+    private Map<String, String> options;
+
+    private Builder() {
+      load = new LoadDescriptor();
+      options = new HashMap<>();
+    }
+
+    public Builder table(TableIdentifier tableIdentifier) {
+      load.setTable(tableIdentifier);
+      return this;
+    }
+
+    public Builder overwrite(boolean isOverwrite) {
+      load.setOverwrite(isOverwrite);
+      return this;
+    }
+
+    public Builder inputPath(String inputPath) {
+      load.setInputPath(inputPath);
+      return this;
+    }
+
+    public Builder options(String key, String value) {
+      options.put(key, value);
+      return this;
+    }
+
+    public LoadDescriptor create() {
+      load.setOptions(options);
+      return load;
+    }
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/SelectDescriptor.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/SelectDescriptor.java b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/SelectDescriptor.java
new file mode 100644
index 0000000..c3627a9
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/SelectDescriptor.java
@@ -0,0 +1,111 @@
+/*
+ * 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.store.api.descriptor;
+
+import java.util.Objects;
+
+import org.apache.carbondata.core.scan.expression.Expression;
+
+public class SelectDescriptor {
+
+  private TableIdentifier table;
+  private String[] projection;
+  private Expression filter;
+  private long limit;
+
+  private SelectDescriptor() {
+  }
+
+  public SelectDescriptor(TableIdentifier table, String[] projection,
+      Expression filter, long limit) {
+    Objects.requireNonNull(table);
+    Objects.requireNonNull(projection);
+    this.table = table;
+    this.projection = projection;
+    this.filter = filter;
+    this.limit = limit;
+  }
+
+  public TableIdentifier getTable() {
+    return table;
+  }
+
+  public void setTable(TableIdentifier table) {
+    this.table = table;
+  }
+
+  public String[] getProjection() {
+    return projection;
+  }
+
+  public void setProjection(String[] projection) {
+    this.projection = projection;
+  }
+
+  public Expression getFilter() {
+    return filter;
+  }
+
+  public void setFilter(Expression filter) {
+    this.filter = filter;
+  }
+
+  public long getLimit() {
+    return limit;
+  }
+
+  public void setLimit(long limit) {
+    this.limit = limit;
+  }
+
+  public static class Builder {
+    private SelectDescriptor select;
+
+    private Builder() {
+      select = new SelectDescriptor();
+    }
+
+    public Builder table(TableIdentifier tableIdentifier) {
+      select.setTable(tableIdentifier);
+      return this;
+    }
+
+    public Builder select(String... columnNames) {
+      select.setProjection(columnNames);
+      return this;
+    }
+
+    public Builder filter(Expression filter) {
+      select.setFilter(filter);
+      return this;
+    }
+
+    public Builder limit(long limit) {
+      select.setLimit(limit);
+      return this;
+    }
+
+    public SelectDescriptor create() {
+      return select;
+    }
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableDescriptor.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableDescriptor.java b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableDescriptor.java
new file mode 100644
index 0000000..2d677a8
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableDescriptor.java
@@ -0,0 +1,174 @@
+/*
+ * 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.store.api.descriptor;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.sdk.file.Field;
+import org.apache.carbondata.sdk.file.Schema;
+
+public class TableDescriptor {
+
+  private boolean ifNotExists;
+  private TableIdentifier table;
+  private String tablePath;
+  private Schema schema;
+  private Map<String, String> properties;
+  private String comment;
+
+  private TableDescriptor() {
+  }
+
+  public TableDescriptor(TableIdentifier table, Schema schema,
+      Map<String, String> properties, String tablePath, String comment, boolean ifNotExists) {
+    Objects.requireNonNull(table);
+    Objects.requireNonNull(schema);
+    this.table = table;
+    this.ifNotExists = ifNotExists;
+    this.schema = schema;
+    this.properties = properties;
+    this.tablePath = tablePath;
+    this.comment = comment;
+  }
+
+  public boolean isIfNotExists() {
+    return ifNotExists;
+  }
+
+  public void setIfNotExists(boolean ifNotExists) {
+    this.ifNotExists = ifNotExists;
+  }
+
+  public TableIdentifier getTable() {
+    return table;
+  }
+
+  public void setTable(TableIdentifier table) {
+    this.table = table;
+  }
+
+  public Schema getSchema() {
+    return schema;
+  }
+
+  public void setSchema(Schema schema) {
+    this.schema = schema;
+  }
+
+  public Map<String, String> getProperties() {
+    return properties;
+  }
+
+  public void setProperties(Map<String, String> properties) {
+    this.properties = properties;
+  }
+
+  public String getComment() {
+    return comment;
+  }
+
+  public void setComment(String comment) {
+    this.comment = comment;
+  }
+
+  public void setTablePath(String tablePath) {
+    this.tablePath = tablePath;
+  }
+
+  public String getTablePath() {
+    return tablePath;
+  }
+
+  public static class Builder {
+
+    private TableDescriptor table;
+    private List<Field> fields;
+    private Map<String, String> tblProperties;
+
+    private Builder() {
+      table = new TableDescriptor();
+      fields = new ArrayList<>();
+      tblProperties = new HashMap<>();
+    }
+
+    public Builder ifNotExists() {
+      table.setIfNotExists(true);
+      return this;
+    }
+
+    public Builder table(TableIdentifier tableId) {
+      table.setTable(tableId);
+      return this;
+    }
+
+    public Builder tablePath(String tablePath) {
+      table.setTablePath(tablePath);
+      return this;
+    }
+
+    public Builder comment(String tableComment) {
+      table.setComment(tableComment);
+      return this;
+    }
+
+    public Builder column(String name, DataType dataType) {
+      fields.add(new Field(name, dataType));
+      return this;
+    }
+
+    public Builder column(String name, DataType dataType, String comment) {
+      Field field = new Field(name, dataType);
+      field.setColumnComment(comment);
+      fields.add(field);
+      return this;
+    }
+
+    public Builder column(String name, DataType dataType, int precision, int scale, String comment)
+    {
+      Field field = new Field(name, dataType);
+      field.setColumnComment(comment);
+      field.setScale(scale);
+      field.setPrecision(precision);
+      fields.add(field);
+      return this;
+    }
+
+    public Builder tblProperties(String key, String value) {
+      tblProperties.put(key, value);
+      return this;
+    }
+
+    public TableDescriptor create() {
+      Field[] fieldArray = new Field[fields.size()];
+      fieldArray = fields.toArray(fieldArray);
+      Schema schema = new Schema(fieldArray);
+      table.setSchema(schema);
+      table.setProperties(tblProperties);
+      return table;
+    }
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableIdentifier.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableIdentifier.java b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableIdentifier.java
new file mode 100644
index 0000000..ab8edf8
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/descriptor/TableIdentifier.java
@@ -0,0 +1,37 @@
+/*
+ * 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.store.api.descriptor;
+
+public class TableIdentifier {
+  private String tableName;
+  private String databaseName;
+
+  public TableIdentifier(String tableName, String databaseName) {
+    this.tableName = tableName;
+    this.databaseName = databaseName;
+  }
+
+  public String getTableName() {
+    return tableName;
+  }
+
+  public String getDatabaseName() {
+    return databaseName;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/exception/ExecutionTimeoutException.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/exception/ExecutionTimeoutException.java b/store/core/src/main/java/org/apache/carbondata/store/api/exception/ExecutionTimeoutException.java
new file mode 100644
index 0000000..728837d
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/exception/ExecutionTimeoutException.java
@@ -0,0 +1,22 @@
+/*
+ * 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.store.api.exception;
+
+public class ExecutionTimeoutException extends RuntimeException {
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/exception/SchedulerException.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/exception/SchedulerException.java b/store/core/src/main/java/org/apache/carbondata/store/api/exception/SchedulerException.java
new file mode 100644
index 0000000..28b8a50
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/exception/SchedulerException.java
@@ -0,0 +1,26 @@
+/*
+ * 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.store.api.exception;
+
+public class SchedulerException extends RuntimeException {
+
+  public SchedulerException(String message) {
+    super(message);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/044a995a/store/core/src/main/java/org/apache/carbondata/store/api/exception/StoreException.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/api/exception/StoreException.java b/store/core/src/main/java/org/apache/carbondata/store/api/exception/StoreException.java
new file mode 100644
index 0000000..315a09b
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/api/exception/StoreException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.store.api.exception;
+
+public class StoreException extends Exception {
+
+  public StoreException() {
+    super();
+  }
+
+  public StoreException(String message) {
+    super(message);
+  }
+
+  public StoreException(Exception e) {
+    super(e);
+  }
+}


Mime
View raw message