carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [4/4] carbondata git commit: [CARBONDATA-2690][CarbonStore] implement RESTful API: create table, load data and select
Date Fri, 06 Jul 2018 07:35:21 GMT
[CARBONDATA-2690][CarbonStore] implement RESTful API: create table, load data and select

This PR adds:
1.basic framework
rewrite the carbon store's Master, Worker and Scheduler code in Java

2.RESTful API
support create a table by using file meta store
support load data to a table in single work
support select data with a filter

This closes #2440


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

Branch: refs/heads/carbonstore
Commit: 6fa86381f82d5907e366ef6707d731a228ac4081
Parents: fa11138
Author: QiangCai <qiangcai@qq.com>
Authored: Tue Jul 3 20:21:18 2018 +0800
Committer: Jacky Li <jacky.likun@qq.com>
Committed: Fri Jul 6 15:34:39 2018 +0800

----------------------------------------------------------------------
 .../schema/table/CarbonTableBuilder.java        |    9 +-
 .../schema/table/TableSchemaBuilder.java        |   23 +-
 .../expression/conditional/ListExpression.java  |    8 +-
 .../conditional/NotEqualsExpression.java        |    7 +-
 .../scan/expression/logical/OrExpression.java   |    2 +-
 .../expression/logical/RangeExpression.java     |    2 +-
 .../carbondata/core/scan/model/QueryModel.java  |    3 +-
 dev/findbugs-exclude.xml                        |    4 +
 dev/javastyle-suppressions.xml                  |    2 +
 .../carbondata/hadoop/CarbonRecordReader.java   |    2 +
 .../hadoop/util/CarbonInputFormatUtil.java      |   20 +-
 .../preaggregate/TestPreAggregateDrop.scala     |    3 +-
 .../TestPreAggregateTableSelection.scala        |    8 +-
 .../carbondata/store/SparkCarbonStore.scala     |   40 +-
 pom.xml                                         |    6 +
 .../loading/model/CarbonLoadModelBuilder.java   |   16 +-
 .../processing/util/CarbonLoaderUtil.java       |    2 +-
 store/conf/log4j.properties                     |   10 +
 store/conf/store.conf                           |   10 +
 store/core/pom.xml                              |    8 +-
 .../apache/carbondata/store/conf/StoreConf.java |  185 +++
 .../exception/ExecutionTimeoutException.java    |   22 +
 .../store/exception/StoreException.java         |   29 +
 .../store/exception/WorkerTooBusyException.java |   26 +
 .../apache/carbondata/store/master/Master.java  |  522 ++++++++
 .../carbondata/store/rpc/QueryService.java      |   33 -
 .../carbondata/store/rpc/RegistryService.java   |    4 +-
 .../carbondata/store/rpc/ServiceFactory.java    |    4 +-
 .../carbondata/store/rpc/StoreService.java      |   40 +
 .../store/rpc/impl/IndexedRecordReader.java     |   22 +
 .../store/rpc/impl/QueryServiceImpl.java        |   56 -
 .../store/rpc/impl/RegistryServiceImpl.java     |    4 +-
 .../store/rpc/impl/RequestHandler.java          |  105 +-
 .../store/rpc/impl/StoreServiceImpl.java        |   78 ++
 .../store/rpc/model/BaseResponse.java           |   69 ++
 .../store/rpc/model/LoadDataRequest.java        |   60 +
 .../store/rpc/model/QueryResponse.java          |   21 +-
 .../store/rpc/model/RegisterWorkerRequest.java  |    4 +
 .../carbondata/store/scheduler/Schedulable.java |   74 ++
 .../carbondata/store/scheduler/Scheduler.java   |  136 +++
 .../apache/carbondata/store/util/StoreUtil.java |  132 +++
 .../apache/carbondata/store/worker/Worker.java  |  166 +++
 .../org/apache/carbondata/store/Master.scala    |  283 -----
 .../org/apache/carbondata/store/Scheduler.scala |  147 ---
 .../org/apache/carbondata/store/Worker.scala    |  113 --
 .../carbondata/store/SchedulerSuite.scala       |  155 ---
 store/horizon/pom.xml                           |   95 ++
 store/horizon/src/main/anltr/Expression.g4      |  163 +++
 .../horizon/antlr/ANTLRNoCaseStringStream.java  |   38 +
 .../carbondata/horizon/antlr/FilterVisitor.java |  215 ++++
 .../horizon/antlr/gen/Expression.tokens         |   51 +
 .../antlr/gen/ExpressionBaseVisitor.java        |  168 +++
 .../horizon/antlr/gen/ExpressionLexer.java      |  228 ++++
 .../horizon/antlr/gen/ExpressionLexer.tokens    |   51 +
 .../horizon/antlr/gen/ExpressionParser.java     | 1117 ++++++++++++++++++
 .../horizon/antlr/gen/ExpressionVisitor.java    |  162 +++
 .../horizon/rest/controller/Horizon.java        |   36 +
 .../rest/controller/HorizonController.java      |   92 ++
 .../rest/model/descriptor/LoadDescriptor.java   |   81 ++
 .../rest/model/descriptor/SelectDescriptor.java |   88 ++
 .../rest/model/descriptor/TableDescriptor.java  |   90 ++
 .../rest/model/validate/RequestValidator.java   |   70 ++
 .../rest/model/view/CreateTableRequest.java     |  174 +++
 .../horizon/rest/model/view/FieldRequest.java   |  114 ++
 .../horizon/rest/model/view/LoadRequest.java    |  132 +++
 .../horizon/rest/model/view/SelectRequest.java  |  130 ++
 .../horizon/rest/model/view/SelectResponse.java |   49 +
 .../horizon/rest/service/HorizonService.java    |  162 +++
 .../carbondata/horizon/FilterParseTest.java     |  161 +++
 .../apache/carbondata/horizon/HorizonTest.java  |  153 +++
 store/horizon/src/test/resources/data1.csv      |   11 +
 .../carbondata/sdk/file/CSVCarbonWriter.java    |    4 +-
 .../sdk/file/CarbonWriterBuilder.java           |    2 +-
 .../org/apache/carbondata/sdk/file/Schema.java  |   67 ++
 74 files changed, 5716 insertions(+), 863 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTableBuilder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTableBuilder.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTableBuilder.java
index e1d2162..cd27cf1 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTableBuilder.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTableBuilder.java
@@ -59,12 +59,11 @@ public class CarbonTableBuilder {
     return this;
   }
 
-  public CarbonTable build() {
+  public TableInfo buildTableInfo() {
     Objects.requireNonNull(tablePath, "tablePath should not be null");
     Objects.requireNonNull(tableSchema, "tableSchema should not be null");
     Objects.requireNonNull(isTransactionalTable, "Transactional Table should not be null");
 
-
     TableInfo tableInfo = new TableInfo();
     tableInfo.setDatabaseName(databaseName);
     tableInfo.setTableUniqueName(databaseName + "_" + tableName);
@@ -73,6 +72,10 @@ public class CarbonTableBuilder {
     tableInfo.setTransactionalTable(isTransactionalTable);
     tableInfo.setLastUpdatedTime(System.currentTimeMillis());
     tableInfo.setDataMapSchemaList(new ArrayList<DataMapSchema>(0));
-    return CarbonTable.buildFromTableInfo(tableInfo);
+    return tableInfo;
+  }
+
+  public CarbonTable build() {
+    return CarbonTable.buildFromTableInfo(buildTableInfo());
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/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 40f8725..e4b072b 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
@@ -54,12 +54,22 @@ public class TableSchemaBuilder {
 
   private List<ColumnSchema> measures = new LinkedList<>();
 
+  private Map<String, String> tableProperties;
+
   private int blockSize;
 
   private int blockletSize;
 
   private String tableName;
 
+  public TableSchemaBuilder properties(Map<String, String> tableProperties) {
+    if (tableProperties == null) {
+      throw new IllegalArgumentException("blockSize should not be null");
+    }
+    this.tableProperties = tableProperties;
+    return this;
+  }
+
   public TableSchemaBuilder blockSize(int blockSize) {
     if (blockSize <= 0) {
       throw new IllegalArgumentException("blockSize should be greater than 0");
@@ -97,15 +107,18 @@ public class TableSchemaBuilder {
     allColumns.addAll(measures);
     schema.setListOfColumns(allColumns);
 
-    Map<String, String> property = new HashMap<>();
+    if (tableProperties == null) {
+      tableProperties = new HashMap<>();
+    }
     if (blockSize > 0) {
-      property.put(CarbonCommonConstants.TABLE_BLOCKSIZE, String.valueOf(blockSize));
+      tableProperties.put(CarbonCommonConstants.TABLE_BLOCKSIZE, String.valueOf(blockSize));
     }
     if (blockletSize > 0) {
-      property.put(CarbonV3DataFormatConstants.BLOCKLET_SIZE_IN_MB, String.valueOf(blockletSize));
+      tableProperties.put(
+          CarbonV3DataFormatConstants.BLOCKLET_SIZE_IN_MB, String.valueOf(blockletSize));
     }
-    if (property.size() != 0) {
-      schema.setTableProperties(property);
+    if (tableProperties.size() != 0) {
+      schema.setTableProperties(tableProperties);
     }
 
     return schema;

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/ListExpression.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/ListExpression.java b/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/ListExpression.java
index 32b5028..a686543 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/ListExpression.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/ListExpression.java
@@ -69,8 +69,14 @@ public class ListExpression extends Expression {
   public String getStatement() {
     StringBuffer value = new StringBuffer();
     value.append("(");
+    boolean start = false;
     for (Expression expr : children) {
-      value.append(expr.getString()).append(";");
+      if (start) {
+        value.append(", ");
+      } else {
+        start = true;
+      }
+      value.append(expr.getStatement());
     }
     value.append(')');
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/NotEqualsExpression.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/NotEqualsExpression.java b/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/NotEqualsExpression.java
index eae8019..438337c 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/NotEqualsExpression.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/NotEqualsExpression.java
@@ -100,6 +100,11 @@ public class NotEqualsExpression extends BinaryConditionalExpression {
 
   @Override
   public String getStatement() {
-    return left.getStatement() + " <> " + right.getStatement();
+    if (isNotNull) {
+      return left.getStatement() + " is not " + right.getStatement();
+    } else {
+      return left.getStatement() + " <> " + right.getStatement();
+    }
+
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/OrExpression.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/OrExpression.java b/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/OrExpression.java
index 148081a..5b7dda0 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/OrExpression.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/OrExpression.java
@@ -59,6 +59,6 @@ public class OrExpression extends BinaryLogicalExpression {
 
   @Override
   public String getStatement() {
-    return "(" + left.getString() + " or " + right.getString() + ")";
+    return "(" + left.getStatement() + " or " + right.getStatement() + ")";
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/RangeExpression.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/RangeExpression.java b/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/RangeExpression.java
index 5d09fb1..127fa50 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/RangeExpression.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/expression/logical/RangeExpression.java
@@ -66,7 +66,7 @@ public class RangeExpression extends BinaryConditionalExpression {
 
   @Override
   public String getStatement() {
-    return left.getStatement() + " between " + right.getStatement();
+    return left.getStatement() + " and " + right.getStatement();
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/core/src/main/java/org/apache/carbondata/core/scan/model/QueryModel.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/model/QueryModel.java b/core/src/main/java/org/apache/carbondata/core/scan/model/QueryModel.java
index b15ce02..72666e3 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/model/QueryModel.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/model/QueryModel.java
@@ -376,6 +376,7 @@ public class QueryModel {
     return String.format("scan on table %s.%s, %d projection columns with filter (%s)",
         table.getDatabaseName(), table.getTableName(),
         projection.getDimensions().size() + projection.getMeasures().size(),
-        filterExpressionResolverTree.getFilterExpression().toString());
+        filterExpressionResolverTree == null ? "" :
+            filterExpressionResolverTree.getFilterExpression().toString());
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/dev/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/dev/findbugs-exclude.xml b/dev/findbugs-exclude.xml
index 63b6bd5..e3f0523 100644
--- a/dev/findbugs-exclude.xml
+++ b/dev/findbugs-exclude.xml
@@ -21,6 +21,10 @@
   </Match>
 
   <Match>
+    <Package name="org.apache.carbondata.horizon.antlr.gen"/>
+  </Match>
+
+  <Match>
     <Source name="~.*\.scala" />
   </Match>
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/dev/javastyle-suppressions.xml
----------------------------------------------------------------------
diff --git a/dev/javastyle-suppressions.xml b/dev/javastyle-suppressions.xml
index 9780dcb..bb4e6f1 100644
--- a/dev/javastyle-suppressions.xml
+++ b/dev/javastyle-suppressions.xml
@@ -32,4 +32,6 @@
               files="org/apache/carbondata/processing/newflow/sort/unsafe/sort/TimSort.java"/>
     <suppress checks=".*"
               files="org/apache/carbondata/core/memory/HeapMemoryAllocator"/>
+    <suppress checks=".*"
+              files="org/apache/carbondata/horizon/antlr/gen"/>
 </suppressions>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java
index 3a0037f..269d740 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java
@@ -118,9 +118,11 @@ public class CarbonRecordReader<T> extends AbstractRecordReader<T> {
         CarbonUtil.clearDictionaryCache(entry.getValue());
       }
     }
+
     // Clear the datamap cache
     DataMapStoreManager.getInstance()
         .clearDataMaps(queryModel.getTable().getAbsoluteTableIdentifier());
+
     // close read support
     readSupport.close();
     try {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
index af7397b..3076cd7 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
@@ -92,6 +92,19 @@ public class CarbonInputFormatUtil {
       Expression filterExpression,
       List<PartitionSpec> partitionNames,
       DataMapJob dataMapJob) throws IOException, InvalidConfigurationException {
+    return createCarbonTableInputFormat(job, carbonTable, projectionColumns, filterExpression,
+        partitionNames, dataMapJob, false);
+  }
+
+
+  public static <V> CarbonTableInputFormat<V> createCarbonTableInputFormat(
+      Job job,
+      CarbonTable carbonTable,
+      String[] projectionColumns,
+      Expression filterExpression,
+      List<PartitionSpec> partitionNames,
+      DataMapJob dataMapJob,
+      boolean isSearchMode) throws IOException, InvalidConfigurationException {
     Configuration conf = job.getConfiguration();
     CarbonInputFormat.setTableInfo(conf, carbonTable.getTableInfo());
     CarbonInputFormat.setDatabaseName(conf, carbonTable.getTableInfo().getDatabaseName());
@@ -103,7 +116,7 @@ public class CarbonInputFormatUtil {
         .setTransactionalTable(conf, carbonTable.getTableInfo().isTransactionalTable());
     CarbonProjection columnProjection = new CarbonProjection(projectionColumns);
     return createInputFormat(conf, carbonTable.getAbsoluteTableIdentifier(),
-        filterExpression, columnProjection, dataMapJob);
+        filterExpression, columnProjection, dataMapJob, isSearchMode);
   }
 
   private static <V> CarbonTableInputFormat<V> createInputFormat(
@@ -111,7 +124,8 @@ public class CarbonInputFormatUtil {
       AbsoluteTableIdentifier identifier,
       Expression filterExpression,
       CarbonProjection columnProjection,
-      DataMapJob dataMapJob) throws InvalidConfigurationException, IOException {
+      DataMapJob dataMapJob,
+      boolean isSearchMode) throws InvalidConfigurationException, IOException {
     CarbonTableInputFormat<V> format = new CarbonTableInputFormat<>();
     CarbonInputFormat.setTablePath(
         conf,
@@ -121,7 +135,7 @@ public class CarbonInputFormatUtil {
     CarbonInputFormat.setColumnProjection(conf, columnProjection);
     if (dataMapJob != null) {
       DataMapUtil.setDataMapJob(conf, dataMapJob);
-    } else {
+    } else if (!isSearchMode) {
       setDataMapJobIfConfigured(conf);
     }
     // when validate segments is disabled in thread local update it to CarbonTableInputFormat

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
index f73a587..2a9fdcd 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
@@ -91,8 +91,7 @@ class TestPreAggregateDrop extends QueryTest with BeforeAndAfterAll {
     sql(
       "create datamap preagg2 on table maintable using 'preaggregate' as select" +
       " a,sum(c) from maintable group by a")
-    sql("drop table if exists maintable")
-    checkExistence(sql("show tables"), false, "maintable_preagg1", "maintable", "maintable_preagg2")
+    checkExistence(sql("show tables").select("database", "tableName"), false, "defaultmaintable_preagg1", "defaultmaintable", "defaultmaintable_preagg2")
   }
 
   test("drop datamap with 'if exists' when datamap not exists") {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
index ebb2491..9e4a0d0 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
@@ -16,8 +16,6 @@
  */
 package org.apache.carbondata.integration.spark.testsuite.preaggregate
 
-import java.io.File
-
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.hive.CarbonRelation
@@ -416,7 +414,7 @@ class TestPreAggregateTableSelection extends SparkQueryTest with BeforeAndAfterA
         | - agg1 (preaggregate)
         |Table Scan on maintable_agg1
         | - total blocklets: 1
-        | - filter: (maintable_name <> null and maintable_name = a)
+        | - filter: (maintable_name is not null and maintable_name = a)
         | - pruned by Main DataMap
         |    - skipped blocklets: 1
         |""".stripMargin)(rows(0).getString(0))
@@ -432,14 +430,14 @@ class TestPreAggregateTableSelection extends SparkQueryTest with BeforeAndAfterA
       """
         |Table Scan on maintable
         | - total blocklets: 1
-        | - filter: ((id <> null and id < 3) and name <> null)
+        | - filter: ((id is not null and id < 3) and name is not null)
         | - pruned by Main DataMap
         |    - skipped blocklets: 0""".stripMargin))
     assert(rows(0).getString(0).contains(
       """
         |Table Scan on maintableavg
         | - total blocklets: 1
-        | - filter: name <> null
+        | - filter: name is not null
         | - pruned by Main DataMap
         |    - skipped blocklets: 0""".stripMargin))
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/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
index d99081d..a4124a2 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/store/SparkCarbonStore.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/store/SparkCarbonStore.scala
@@ -22,7 +22,7 @@ import java.net.InetAddress
 
 import scala.collection.JavaConverters._
 
-import org.apache.spark.{CarbonInputMetrics, SparkConf}
+import org.apache.spark.{CarbonInputMetrics, SparkConf, SparkEnv}
 import org.apache.spark.sql.CarbonSession._
 import org.apache.spark.sql.SparkSession
 
@@ -34,6 +34,10 @@ 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
@@ -110,7 +114,11 @@ class SparkCarbonStore extends MetaCachedCarbonStore {
 
   def startSearchMode(): Unit = {
     LOG.info("Starting search mode master")
-    master = new 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()
   }
@@ -154,7 +162,33 @@ class SparkCarbonStore extends MetaCachedCarbonStore {
     val rows = session.sparkContext.parallelize(1 to numExecutors * 10, numExecutors)
       .mapPartitions { f =>
         // start worker
-        Worker.init(masterIp, CarbonProperties.getSearchMasterPort)
+        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
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 5e81f2b..bbf6c9d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -583,6 +583,12 @@
       </build>
     </profile>
     <profile>
+      <id>horizon</id>
+      <modules>
+        <module>store/horizon</module>
+      </modules>
+    </profile>
+    <profile>
       <id>include-all</id>
     </profile>
     <profile>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java b/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java
index a88ce60..fa54f1b 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/model/CarbonLoadModelBuilder.java
@@ -50,10 +50,20 @@ public class CarbonLoadModelBuilder {
 
   private CarbonTable table;
 
+  private String inputPath;
+
   public CarbonLoadModelBuilder(CarbonTable table) {
     this.table = table;
   }
 
+  public String getInputPath() {
+    return inputPath;
+  }
+
+  public void setInputPath(String inputPath) {
+    this.inputPath = inputPath;
+  }
+
   /**
    * build CarbonLoadModel for data loading
    * @param options Load options from user input
@@ -64,7 +74,9 @@ public class CarbonLoadModelBuilder {
       throws InvalidLoadOptionException, IOException {
     Map<String, String> optionsFinal = LoadOption.fillOptionWithDefaultValue(options);
 
-    if (!options.containsKey("fileheader")) {
+    if (inputPath == null &&
+        !options.containsKey("header") &&
+        !options.containsKey("fileheader")) {
       List<CarbonColumn> csvHeader = table.getCreateOrderColumn(table.getTableName());
       String[] columns = new String[csvHeader.size()];
       for (int i = 0; i < columns.length; i++) {
@@ -72,10 +84,12 @@ public class CarbonLoadModelBuilder {
       }
       optionsFinal.put("fileheader", Strings.mkString(columns, ","));
     }
+
     CarbonLoadModel model = new CarbonLoadModel();
     model.setCarbonTransactionalTable(table.isTransactionalTable());
     model.setFactTimeStamp(UUID);
     model.setTaskNo(taskNo);
+    model.setFactFilePath(inputPath);
 
     // we have provided 'fileheader', so it hadoopConf can be null
     build(options, optionsFinal, model, null);

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/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 da77cf6..29ce2f4 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
@@ -577,7 +577,7 @@ public final class CarbonLoaderUtil {
    * @param noOfNodesInput -1 if number of nodes has to be decided
    *                       based on block location information
    * @param blockAssignmentStrategy strategy used to assign blocks
-   * @param loadMinSize the property load_min_size_inmb specified by the user
+   * @param expectedMinSizePerNode expected minimum size per node
    * @return a map that maps node to blocks
    */
   public static Map<String, List<Distributable>> nodeBlockMapping(

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/conf/log4j.properties
----------------------------------------------------------------------
diff --git a/store/conf/log4j.properties b/store/conf/log4j.properties
new file mode 100755
index 0000000..101ef37
--- /dev/null
+++ b/store/conf/log4j.properties
@@ -0,0 +1,10 @@
+log4j.rootLogger=INFO, C1, F1
+log4j.appender.C1=org.apache.log4j.ConsoleAppender
+log4j.appender.C1.layout=org.apache.log4j.PatternLayout
+log4j.appender.C1.layout.ConversionPattern=%d %p %m%n
+
+log4j.appender.F1=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.F1.file=${log.path}
+log4j.appender.F1.append=true
+log4j.appender.F1.layout=org.apache.log4j.PatternLayout
+log4j.appender.F1.layout.ConversionPattern=%d %p %m%n

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/conf/store.conf
----------------------------------------------------------------------
diff --git a/store/conf/store.conf b/store/conf/store.conf
new file mode 100644
index 0000000..7f18076
--- /dev/null
+++ b/store/conf/store.conf
@@ -0,0 +1,10 @@
+# worker
+carbon.worker.host=127.0.0.1
+carbon.worker.port=10021
+carbon.worker.core.num=2
+carbon.store.temp.location=/tmp/carbon.store.temp
+
+# worker and master
+carbon.master.host=127.0.0.1
+carbon.master.port=10020
+carbon.store.location=/tmp/carbon.store
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/pom.xml
----------------------------------------------------------------------
diff --git a/store/core/pom.xml b/store/core/pom.xml
index 0bee84f..6b2703e 100644
--- a/store/core/pom.xml
+++ b/store/core/pom.xml
@@ -16,19 +16,14 @@
 
   <properties>
     <dev.path>${basedir}/../../dev</dev.path>
+    <spring.version>2.0.2.RELEASE</spring.version>
   </properties>
 
   <dependencies>
     <dependency>
       <groupId>org.apache.carbondata</groupId>
-      <artifactId>carbondata-hadoop</artifactId>
-      <version>${project.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.carbondata</groupId>
       <artifactId>carbondata-store-sdk</artifactId>
       <version>${project.version}</version>
-      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>junit</groupId>
@@ -43,7 +38,6 @@
   </dependencies>
 
   <build>
-    <testSourceDirectory>src/test/scala</testSourceDirectory>
     <plugins>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/conf/StoreConf.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/conf/StoreConf.java b/store/core/src/main/java/org/apache/carbondata/store/conf/StoreConf.java
new file mode 100644
index 0000000..da2a697
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/conf/StoreConf.java
@@ -0,0 +1,185 @@
+/*
+ * 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.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";
+
+  private Map<String, String> conf = new HashMap<>();
+
+  public StoreConf() {
+  }
+
+  public StoreConf(String filePath) {
+    load(filePath);
+  }
+
+  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/6fa86381/store/core/src/main/java/org/apache/carbondata/store/exception/ExecutionTimeoutException.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/exception/ExecutionTimeoutException.java b/store/core/src/main/java/org/apache/carbondata/store/exception/ExecutionTimeoutException.java
new file mode 100644
index 0000000..c7a4d6b
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/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.exception;
+
+public class ExecutionTimeoutException extends RuntimeException {
+
+}

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

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/exception/WorkerTooBusyException.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/exception/WorkerTooBusyException.java b/store/core/src/main/java/org/apache/carbondata/store/exception/WorkerTooBusyException.java
new file mode 100644
index 0000000..b366a67
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/exception/WorkerTooBusyException.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.exception;
+
+public class WorkerTooBusyException extends RuntimeException {
+
+  public WorkerTooBusyException(String message) {
+    super(message);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/master/Master.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/master/Master.java b/store/core/src/main/java/org/apache/carbondata/store/master/Master.java
new file mode 100644
index 0000000..0a724d9
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/master/Master.java
@@ -0,0 +1,522 @@
+/*
+ * 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.master;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.ref.SoftReference;
+import java.net.BindException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.block.Distributable;
+import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.datastore.row.CarbonRow;
+import org.apache.carbondata.core.exception.InvalidConfigurationException;
+import org.apache.carbondata.core.fileoperations.FileWriteOperation;
+import org.apache.carbondata.core.locks.CarbonLockUtil;
+import org.apache.carbondata.core.locks.ICarbonLock;
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.SegmentFileStore;
+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.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.scan.expression.Expression;
+import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
+import org.apache.carbondata.core.writer.ThriftWriter;
+import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
+import org.apache.carbondata.hadoop.api.CarbonInputFormat;
+import org.apache.carbondata.hadoop.api.CarbonTableInputFormat;
+import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil;
+import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
+import org.apache.carbondata.processing.util.CarbonLoaderUtil;
+import org.apache.carbondata.store.conf.StoreConf;
+import org.apache.carbondata.store.exception.ExecutionTimeoutException;
+import org.apache.carbondata.store.exception.StoreException;
+import org.apache.carbondata.store.rpc.RegistryService;
+import org.apache.carbondata.store.rpc.ServiceFactory;
+import org.apache.carbondata.store.rpc.StoreService;
+import org.apache.carbondata.store.rpc.impl.RegistryServiceImpl;
+import org.apache.carbondata.store.rpc.impl.Status;
+import org.apache.carbondata.store.rpc.model.BaseResponse;
+import org.apache.carbondata.store.rpc.model.LoadDataRequest;
+import org.apache.carbondata.store.rpc.model.QueryRequest;
+import org.apache.carbondata.store.rpc.model.QueryResponse;
+import org.apache.carbondata.store.rpc.model.RegisterWorkerRequest;
+import org.apache.carbondata.store.rpc.model.RegisterWorkerResponse;
+import org.apache.carbondata.store.rpc.model.ShutdownRequest;
+import org.apache.carbondata.store.scheduler.Schedulable;
+import org.apache.carbondata.store.scheduler.Scheduler;
+import org.apache.carbondata.store.util.StoreUtil;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+
+/**
+ * Master of CarbonSearch.
+ * It provides a Registry service for worker to register.
+ * And it provides search API to fire RPC call to workers.
+ */
+
+public class Master {
+
+  private static Master instance = null;
+
+  private static LogService LOGGER = LogServiceFactory.getLogService(Master.class.getName());
+
+  private Map<String, SoftReference<CarbonTable>> cacheTables;
+
+  // worker host address map to EndpointRef
+  private StoreConf conf;
+  private Configuration hadoopConf;
+  private Random random = new Random();
+  private RPC.Server registryServer = null;
+  private Scheduler scheduler = new Scheduler();
+
+  private Master(StoreConf conf) {
+    cacheTables = new HashMap<>();
+    this.conf = conf;
+    this.hadoopConf = this.conf.newHadoopConf();
+  }
+
+  /**
+   * start service and listen on port passed in constructor
+   */
+  public void startService() throws IOException {
+    if (registryServer == null) {
+
+      BindException exception;
+      // we will try to create service at worse case 100 times
+      int numTry = 100;
+      String host = conf.masterHost();
+      int port = conf.masterPort();
+      LOGGER.info("building registry-service on " + host + ":" + port);
+
+      RegistryService registryService = new RegistryServiceImpl(this);
+      do {
+        try {
+          registryServer = new RPC.Builder(hadoopConf).setBindAddress(host).setPort(port)
+              .setProtocol(RegistryService.class).setInstance(registryService).build();
+
+          registryServer.start();
+          numTry = 0;
+          exception = null;
+        } catch (BindException e) {
+          // port is occupied, increase the port number and try again
+          exception = e;
+          LOGGER.error(e, "start registry-service failed");
+          port = port + 1;
+          numTry = numTry - 1;
+        }
+      } while (numTry > 0);
+      if (exception != null) {
+        // we have tried many times, but still failed to find an available port
+        throw exception;
+      }
+      LOGGER.info("registry-service started");
+    } else {
+      LOGGER.info("Search mode master has already started");
+    }
+  }
+
+  public void stopService() throws InterruptedException {
+    if (registryServer != null) {
+      registryServer.stop();
+      registryServer.join();
+      registryServer = null;
+    }
+  }
+
+  public void stopAllWorkers() throws IOException {
+    for (Schedulable worker : getWorkers()) {
+      try {
+        worker.service.shutdown(new ShutdownRequest("user"));
+      } catch (Throwable throwable) {
+        throw new IOException(throwable);
+      }
+      scheduler.removeWorker(worker.getAddress());
+    }
+  }
+
+  /**
+   * A new searcher is trying to register, add it to the map and connect to this searcher
+   */
+  public RegisterWorkerResponse addWorker(RegisterWorkerRequest request) throws IOException {
+    LOGGER.info(
+        "Receive Register request from worker " + request.getHostAddress() + ":" + request.getPort()
+            + " with " + request.getCores() + " cores");
+    String workerId = UUID.randomUUID().toString();
+    String workerAddress = request.getHostAddress();
+    int workerPort = request.getPort();
+    LOGGER.info(
+        "connecting to worker " + request.getHostAddress() + ":" + request.getPort() + ", workerId "
+            + workerId);
+
+    StoreService searchService = ServiceFactory.createStoreService(workerAddress, workerPort);
+    scheduler.addWorker(
+        new Schedulable(workerId, workerAddress, workerPort, request.getCores(), searchService));
+    LOGGER.info("worker " + request + " registered");
+    return new RegisterWorkerResponse(workerId);
+  }
+
+  private int onSuccess(int queryId, QueryResponse result, List<CarbonRow> output, long globalLimit)
+      throws IOException {
+    // in case of RPC success, collect all rows in response message
+    if (result.getQueryId() != queryId) {
+      throw new IOException(
+          "queryId in response does not match request: " + result.getQueryId() + " != " + queryId);
+    }
+    if (result.getStatus() != Status.SUCCESS.ordinal()) {
+      throw new IOException("failure in worker: " + result.getMessage());
+    }
+    int rowCount = 0;
+    Object[][] rows = result.getRows();
+    for (Object[] row : rows) {
+      output.add(new CarbonRow(row));
+      rowCount++;
+      if (rowCount >= globalLimit) {
+        break;
+      }
+    }
+    LOGGER.info("[QueryId:" + queryId + "] accumulated result size " + rowCount);
+    return rowCount;
+  }
+
+  private void onFailure(Throwable e) throws IOException {
+    throw new IOException("exception in worker: " + e.getMessage());
+  }
+
+  private void onTimeout() {
+    throw new ExecutionTimeoutException();
+  }
+
+  public String getTableFolder(String database, String tableName) {
+    return conf.storeLocation() + File.separator + database + File.separator + tableName;
+  }
+
+  public CarbonTable getTable(String database, String tableName) throws StoreException {
+    String tablePath = getTableFolder(database, tableName);
+    CarbonTable carbonTable;
+    SoftReference<CarbonTable> reference = cacheTables.get(tablePath);
+    if (reference != null) {
+      carbonTable = reference.get();
+      if (carbonTable != null) {
+        return carbonTable;
+      }
+    }
+
+    try {
+      org.apache.carbondata.format.TableInfo tableInfo =
+          CarbonUtil.readSchemaFile(CarbonTablePath.getSchemaFilePath(tablePath));
+      SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
+      TableInfo tableInfo1 = schemaConverter.fromExternalToWrapperTableInfo(tableInfo, "", "", "");
+      tableInfo1.setTablePath(tablePath);
+      carbonTable = CarbonTable.buildFromTableInfo(tableInfo1);
+      cacheTables.put(tablePath, new SoftReference<>(carbonTable));
+      return carbonTable;
+    } catch (IOException e) {
+      String message = "Failed to get table from " + tablePath;
+      LOGGER.error(e, message);
+      throw new StoreException(message);
+    }
+  }
+
+  public boolean createTable(TableInfo tableInfo, boolean ifNotExists) throws IOException {
+    AbsoluteTableIdentifier identifier = tableInfo.getOrCreateAbsoluteTableIdentifier();
+    boolean tableExists = FileFactory.isFileExist(identifier.getTablePath());
+    if (tableExists) {
+      if (ifNotExists) {
+        return true;
+      } else {
+        throw new IOException(
+            "car't create table " + tableInfo.getDatabaseName() + "." + tableInfo.getFactTable()
+                .getTableName() + ", because it already exists");
+      }
+    }
+
+    SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
+    String databaseName = tableInfo.getDatabaseName();
+    String tableName = tableInfo.getFactTable().getTableName();
+    org.apache.carbondata.format.TableInfo thriftTableInfo =
+        schemaConverter.fromWrapperToExternalTableInfo(tableInfo, databaseName, tableName);
+
+    String schemaFilePath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath());
+    String schemaMetadataPath = CarbonTablePath.getFolderContainingFile(schemaFilePath);
+    FileFactory.FileType fileType = FileFactory.getFileType(schemaMetadataPath);
+    try {
+      if (!FileFactory.isFileExist(schemaMetadataPath, fileType)) {
+        boolean isDirCreated = FileFactory.mkdirs(schemaMetadataPath, fileType);
+        if (!isDirCreated) {
+          throw new IOException("Failed to create the metadata directory " + schemaMetadataPath);
+        }
+      }
+      ThriftWriter thriftWriter = new ThriftWriter(schemaFilePath, false);
+      thriftWriter.open(FileWriteOperation.OVERWRITE);
+      thriftWriter.write(thriftTableInfo);
+      thriftWriter.close();
+      return true;
+    } catch (IOException e) {
+      LOGGER.error(e, "Failed to handle create table");
+      throw e;
+    }
+  }
+
+  private void openSegment(CarbonLoadModel loadModel, boolean isOverwriteTable) throws IOException {
+    try {
+      CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(loadModel, isOverwriteTable);
+    } catch (IOException e) {
+      LOGGER.error(e, "Failed to handle load data");
+      throw e;
+    }
+  }
+
+  private void closeSegment(CarbonLoadModel loadModel) throws IOException {
+    try {
+      CarbonLoaderUtil.updateTableStatusForFailure(loadModel, "");
+    } catch (IOException e) {
+      LOGGER.error(e, "Failed to close segment");
+      throw e;
+    }
+  }
+
+  private void commitSegment(CarbonLoadModel loadModel) throws IOException {
+    CarbonTable carbonTable = loadModel.getCarbonDataLoadSchema().getCarbonTable();
+    String segmentId = loadModel.getSegmentId();
+    String segmentFileName = SegmentFileStore
+        .writeSegmentFile(carbonTable, segmentId, String.valueOf(loadModel.getFactTimeStamp()));
+
+    AbsoluteTableIdentifier absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier();
+    String tablePath = absoluteTableIdentifier.getTablePath();
+    String metadataPath = CarbonTablePath.getMetadataPath(tablePath);
+    String tableStatusPath = CarbonTablePath.getTableStatusFilePath(tablePath);
+
+    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
+    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
+    int retryCount = CarbonLockUtil
+        .getLockProperty(CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK,
+            CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT);
+    int maxTimeout = CarbonLockUtil
+        .getLockProperty(CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
+            CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT);
+    try {
+      if (carbonLock.lockWithRetries(retryCount, maxTimeout)) {
+        LOGGER.info("Acquired lock for tablepath" + tablePath + " for table status updation");
+        LoadMetadataDetails[] listOfLoadFolderDetailsArray =
+            SegmentStatusManager.readLoadMetadata(metadataPath);
+        LoadMetadataDetails loadMetadataDetails = null;
+        for (LoadMetadataDetails detail : listOfLoadFolderDetailsArray) {
+          // if the segments is in the list of marked for delete then update the status.
+          if (segmentId.equals(detail.getLoadName())) {
+            loadMetadataDetails = detail;
+            detail.setSegmentFile(segmentFileName);
+            break;
+          }
+        }
+        if (loadMetadataDetails == null) {
+          throw new IOException("can not find segment: " + segmentId);
+        }
+
+        CarbonLoaderUtil.populateNewLoadMetaEntry(loadMetadataDetails, SegmentStatus.SUCCESS,
+            loadModel.getFactTimeStamp(), true);
+        CarbonLoaderUtil
+            .addDataIndexSizeIntoMetaEntry(loadMetadataDetails, segmentId, carbonTable);
+
+        SegmentStatusManager
+            .writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray);
+      } else {
+        LOGGER.error(
+            "Not able to acquire the lock for Table status updation for table path " + tablePath);
+      }
+    } finally {
+      if (carbonLock.unlock()) {
+        LOGGER.info("Table unlocked successfully after table status updation" + tablePath);
+      } else {
+        LOGGER.error(
+            "Unable to unlock Table lock for table" + tablePath + " during table status updation");
+      }
+    }
+  }
+
+  public boolean loadData(CarbonLoadModel loadModel, boolean isOverwrite) throws IOException {
+    Schedulable worker = scheduler.pickNexWorker();
+    try {
+      if (loadModel.getFactTimeStamp() == 0) {
+        loadModel.setFactTimeStamp(CarbonUpdateUtil.readCurrentTime());
+      }
+      openSegment(loadModel, isOverwrite);
+      LoadDataRequest request = new LoadDataRequest(loadModel);
+      BaseResponse response = scheduler.sendRequest(worker, request);
+      if (Status.SUCCESS.ordinal() == response.getStatus()) {
+        commitSegment(loadModel);
+        return true;
+      } else {
+        closeSegment(loadModel);
+        throw new IOException(response.getMessage());
+      }
+    } finally {
+      worker.workload.decrementAndGet();
+    }
+  }
+
+  /**
+   * Execute search by firing RPC call to worker, return the result rows
+   *
+   * @param table       table to search
+   * @param columns     projection column names
+   * @param filter      filter expression
+   * @param globalLimit max number of rows required in Master
+   * @param localLimit  max number of rows required in Worker
+   * @return CarbonRow array
+   */
+  public CarbonRow[] search(CarbonTable table, String[] columns, Expression filter,
+      long globalLimit, long localLimit) throws IOException {
+    Objects.requireNonNull(table);
+    Objects.requireNonNull(columns);
+    if (globalLimit < 0 || localLimit < 0) {
+      throw new IllegalArgumentException("limit should be positive");
+    }
+
+    int queryId = random.nextInt();
+
+    List<CarbonRow> output = new ArrayList<>();
+
+    // prune data and get a mapping of worker hostname to list of blocks,
+    // then add these blocks to the QueryRequest and fire the RPC call
+    Map<String, List<Distributable>> nodeBlockMapping = pruneBlock(table, columns, filter);
+    Set<Map.Entry<String, List<Distributable>>> entries = nodeBlockMapping.entrySet();
+    List<Future<QueryResponse>> futures = new ArrayList<>(entries.size());
+    List<Schedulable> workers = new ArrayList<>(entries.size());
+    for (Map.Entry<String, List<Distributable>> entry : entries) {
+      CarbonMultiBlockSplit split = new CarbonMultiBlockSplit(entry.getValue(), entry.getKey());
+      QueryRequest request =
+          new QueryRequest(queryId, split, table.getTableInfo(), columns, filter, localLimit);
+
+      // 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
+      Schedulable worker = scheduler.pickWorker(entry.getKey());
+      workers.add(worker);
+      futures.add(scheduler.sendRequestAsync(worker, request));
+    }
+
+    int rowCount = 0;
+    int length = futures.size();
+    for (int i = 0; i < length; i++) {
+      Future<QueryResponse> future = futures.get(i);
+      Schedulable worker = workers.get(i);
+      if (rowCount < globalLimit) {
+        // wait for worker
+        QueryResponse response = null;
+        try {
+          response = future
+              .get((long) (CarbonProperties.getInstance().getQueryTimeout()), TimeUnit.SECONDS);
+        } catch (ExecutionException | InterruptedException e) {
+          onFailure(e);
+        } catch (TimeoutException t) {
+          onTimeout();
+        } finally {
+          worker.workload.decrementAndGet();
+        }
+        LOGGER.info("[QueryId: " + queryId + "] receive search response from worker " + worker);
+        rowCount += onSuccess(queryId, response, output, globalLimit);
+      }
+    }
+    CarbonRow[] rows = new CarbonRow[output.size()];
+    return output.toArray(rows);
+  }
+
+  /**
+   * Prune data by using CarbonInputFormat.getSplit
+   * Return a mapping of host address to list of block
+   */
+  private Map<String, List<Distributable>> pruneBlock(CarbonTable table, String[] columns,
+      Expression filter) throws IOException {
+    JobConf jobConf = new JobConf(new Configuration());
+    Job job = new Job(jobConf);
+    CarbonTableInputFormat format;
+    try {
+      format = CarbonInputFormatUtil
+          .createCarbonTableInputFormat(job, table, columns, filter, null, null, true);
+    } catch (InvalidConfigurationException e) {
+      throw new IOException(e.getMessage());
+    }
+
+    // We will do FG pruning in reader side, so don't do it here
+    CarbonInputFormat.setFgDataMapPruning(job.getConfiguration(), false);
+    List<InputSplit> splits = format.getSplits(job);
+    List<Distributable> blockInfos = new ArrayList<>(splits.size());
+    for (InputSplit split : splits) {
+      blockInfos.add((Distributable) split);
+    }
+    return CarbonLoaderUtil.nodeBlockMapping(blockInfos, -1, getWorkerAddresses(),
+        CarbonLoaderUtil.BlockAssignmentStrategy.BLOCK_NUM_FIRST, null);
+  }
+
+  /**
+   * return hostname of all workers
+   */
+  public List<Schedulable> getWorkers() {
+    return scheduler.getAllWorkers();
+  }
+
+  private List<String> getWorkerAddresses() {
+    return scheduler.getAllWorkerAddresses();
+  }
+
+  public static synchronized Master getInstance(StoreConf conf) {
+    if (instance == null) {
+      instance = new Master(conf);
+    }
+    return instance;
+  }
+
+  public static void main(String[] args) throws InterruptedException {
+    if (args.length != 2) {
+      System.err.println("Usage: Master <log4j file> <properties file>");
+      return;
+    }
+
+    StoreUtil.initLog4j(args[0]);
+    StoreConf conf = new StoreConf(args[1]);
+    Master master = getInstance(conf);
+    master.stopService();
+  }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/QueryService.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/QueryService.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/QueryService.java
deleted file mode 100644
index faaa746..0000000
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/QueryService.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.store.rpc;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.store.rpc.model.QueryRequest;
-import org.apache.carbondata.store.rpc.model.QueryResponse;
-import org.apache.carbondata.store.rpc.model.ShutdownRequest;
-import org.apache.carbondata.store.rpc.model.ShutdownResponse;
-
-import org.apache.hadoop.ipc.VersionedProtocol;
-
-@InterfaceAudience.Internal
-public interface QueryService extends VersionedProtocol {
-  long versionID = 1L;
-  QueryResponse query(QueryRequest request);
-  ShutdownResponse shutdown(ShutdownRequest request);
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/RegistryService.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/RegistryService.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/RegistryService.java
index 4d17686..08a0e97 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/RegistryService.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/RegistryService.java
@@ -17,6 +17,8 @@
 
 package org.apache.carbondata.store.rpc;
 
+import java.io.IOException;
+
 import org.apache.carbondata.common.annotations.InterfaceAudience;
 import org.apache.carbondata.store.rpc.model.RegisterWorkerRequest;
 import org.apache.carbondata.store.rpc.model.RegisterWorkerResponse;
@@ -26,5 +28,5 @@ import org.apache.hadoop.ipc.VersionedProtocol;
 @InterfaceAudience.Internal
 public interface RegistryService extends VersionedProtocol {
   long versionID = 1L;
-  RegisterWorkerResponse registerWorker(RegisterWorkerRequest request);
+  RegisterWorkerResponse registerWorker(RegisterWorkerRequest request) throws IOException;
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/ServiceFactory.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/ServiceFactory.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/ServiceFactory.java
index a50ab8b..d9d0f3e 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/ServiceFactory.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/ServiceFactory.java
@@ -29,10 +29,10 @@ import org.apache.hadoop.ipc.RPC;
 @InterfaceAudience.Internal
 public class ServiceFactory {
 
-  public static QueryService createSearchService(String host, int port) throws IOException {
+  public static StoreService createStoreService(String host, int port) throws IOException {
     InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(host), port);
     return RPC.getProxy(
-        QueryService.class, QueryService.versionID, address, new Configuration());
+        StoreService.class, StoreService.versionID, address, new Configuration());
   }
 
   public static RegistryService createRegistryService(String host, int port) throws IOException {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/StoreService.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/StoreService.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/StoreService.java
new file mode 100644
index 0000000..48dec79
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/StoreService.java
@@ -0,0 +1,40 @@
+/*
+ * 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.rpc;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.store.rpc.model.BaseResponse;
+import org.apache.carbondata.store.rpc.model.LoadDataRequest;
+import org.apache.carbondata.store.rpc.model.QueryRequest;
+import org.apache.carbondata.store.rpc.model.QueryResponse;
+import org.apache.carbondata.store.rpc.model.ShutdownRequest;
+import org.apache.carbondata.store.rpc.model.ShutdownResponse;
+
+import org.apache.hadoop.ipc.VersionedProtocol;
+
+@InterfaceAudience.Internal
+public interface StoreService extends VersionedProtocol {
+
+  long versionID = 1L;
+
+  BaseResponse loadData(LoadDataRequest request);
+
+  QueryResponse query(QueryRequest request);
+
+  ShutdownResponse shutdown(ShutdownRequest request);
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/IndexedRecordReader.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/IndexedRecordReader.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/IndexedRecordReader.java
index 2c768d1..45e2dce 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/IndexedRecordReader.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/IndexedRecordReader.java
@@ -21,10 +21,12 @@ import java.io.IOException;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 import org.apache.carbondata.common.logging.LogService;
 import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.dictionary.Dictionary;
 import org.apache.carbondata.core.datamap.DataMapChooser;
 import org.apache.carbondata.core.datamap.DataMapDistributable;
 import org.apache.carbondata.core.datamap.Segment;
@@ -41,6 +43,7 @@ import org.apache.carbondata.core.scan.model.QueryModel;
 import org.apache.carbondata.core.scan.result.iterator.ChunkRowIterator;
 import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
 import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
+import org.apache.carbondata.core.util.CarbonUtil;
 import org.apache.carbondata.core.util.path.CarbonTablePath;
 import org.apache.carbondata.hadoop.CarbonInputSplit;
 import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
@@ -158,4 +161,23 @@ public class IndexedRecordReader extends CarbonRecordReader<CarbonRow> {
     queryModel.setTableBlockInfos(blockToRead);
     return queryModel;
   }
+
+  @Override public void close() throws IOException {
+    logStatistics(rowCount, queryModel.getStatisticsRecorder());
+    // clear dictionary cache
+    Map<String, Dictionary> columnToDictionaryMapping = queryModel.getColumnToDictionaryMapping();
+    if (null != columnToDictionaryMapping) {
+      for (Map.Entry<String, Dictionary> entry : columnToDictionaryMapping.entrySet()) {
+        CarbonUtil.clearDictionaryCache(entry.getValue());
+      }
+    }
+
+    // close read support
+    readSupport.close();
+    try {
+      queryExecutor.finish();
+    } catch (QueryExecutionException e) {
+      throw new IOException(e);
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/QueryServiceImpl.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/QueryServiceImpl.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/QueryServiceImpl.java
deleted file mode 100644
index b191331..0000000
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/QueryServiceImpl.java
+++ /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.carbondata.store.rpc.impl;
-
-import java.io.IOException;
-
-import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.store.rpc.QueryService;
-import org.apache.carbondata.store.rpc.model.QueryRequest;
-import org.apache.carbondata.store.rpc.model.QueryResponse;
-import org.apache.carbondata.store.rpc.model.ShutdownRequest;
-import org.apache.carbondata.store.rpc.model.ShutdownResponse;
-
-import org.apache.hadoop.ipc.ProtocolSignature;
-
-@InterfaceAudience.Internal
-public class QueryServiceImpl implements QueryService {
-
-  @Override
-  public QueryResponse query(QueryRequest request) {
-    RequestHandler handler = new RequestHandler();
-    return handler.handleSearch(request);
-  }
-
-  @Override
-  public ShutdownResponse shutdown(ShutdownRequest request) {
-    RequestHandler handler = new RequestHandler();
-    return handler.handleShutdown(request);
-  }
-
-  @Override
-  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
-    return versionID;
-  }
-
-  @Override
-  public ProtocolSignature getProtocolSignature(String protocol, long clientVersion,
-      int clientMethodsHash) throws IOException {
-    return null;
-  }
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RegistryServiceImpl.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RegistryServiceImpl.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RegistryServiceImpl.java
index 12f48ba..03f9b2c 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RegistryServiceImpl.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RegistryServiceImpl.java
@@ -20,7 +20,7 @@ package org.apache.carbondata.store.rpc.impl;
 import java.io.IOException;
 
 import org.apache.carbondata.common.annotations.InterfaceAudience;
-import org.apache.carbondata.store.Master;
+import org.apache.carbondata.store.master.Master;
 import org.apache.carbondata.store.rpc.RegistryService;
 import org.apache.carbondata.store.rpc.model.RegisterWorkerRequest;
 import org.apache.carbondata.store.rpc.model.RegisterWorkerResponse;
@@ -37,7 +37,7 @@ public class RegistryServiceImpl implements RegistryService {
   }
 
   @Override
-  public RegisterWorkerResponse registerWorker(RegisterWorkerRequest request) {
+  public RegisterWorkerResponse registerWorker(RegisterWorkerRequest request) throws IOException {
     return master.addWorker(request);
   }
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RequestHandler.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RequestHandler.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RequestHandler.java
index 29ee546..3b98019 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RequestHandler.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/RequestHandler.java
@@ -18,10 +18,12 @@
 package org.apache.carbondata.store.rpc.impl;
 
 import java.io.IOException;
+import java.util.Date;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 
+import org.apache.carbondata.common.CarbonIterator;
 import org.apache.carbondata.common.annotations.InterfaceAudience;
 import org.apache.carbondata.common.logging.LogService;
 import org.apache.carbondata.common.logging.LogServiceFactory;
@@ -37,10 +39,29 @@ import org.apache.carbondata.core.util.CarbonTaskInfo;
 import org.apache.carbondata.core.util.ThreadLocalTaskInfo;
 import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
 import org.apache.carbondata.hadoop.CarbonRecordReader;
+import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil;
+import org.apache.carbondata.processing.loading.DataLoadExecutor;
+import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat;
+import org.apache.carbondata.processing.loading.csvinput.CSVRecordReaderIterator;
+import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
+import org.apache.carbondata.store.conf.StoreConf;
+import org.apache.carbondata.store.rpc.model.BaseResponse;
+import org.apache.carbondata.store.rpc.model.LoadDataRequest;
 import org.apache.carbondata.store.rpc.model.QueryRequest;
 import org.apache.carbondata.store.rpc.model.QueryResponse;
 import org.apache.carbondata.store.rpc.model.ShutdownRequest;
 import org.apache.carbondata.store.rpc.model.ShutdownResponse;
+import org.apache.carbondata.store.util.StoreUtil;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 
 /**
  * It handles request from master.
@@ -48,27 +69,35 @@ import org.apache.carbondata.store.rpc.model.ShutdownResponse;
 @InterfaceAudience.Internal
 class RequestHandler {
 
-  private static final LogService LOG =
+  private StoreConf conf;
+  private Configuration hadoopConf;
+
+  public RequestHandler(StoreConf conf, Configuration hadoopConf) {
+    this.conf = conf;
+    this.hadoopConf = hadoopConf;
+  }
+
+  private static final LogService LOGGER =
       LogServiceFactory.getLogService(RequestHandler.class.getName());
 
   QueryResponse handleSearch(QueryRequest request) {
     try {
-      LOG.info(String.format("[QueryId:%d] receive search request", request.getRequestId()));
+      LOGGER.info(String.format("[QueryId:%d] receive search request", request.getRequestId()));
       List<CarbonRow> rows = handleRequest(request);
-      LOG.info(String.format("[QueryId:%d] sending success response", request.getRequestId()));
+      LOGGER.info(String.format("[QueryId:%d] sending success response", request.getRequestId()));
       return createSuccessResponse(request, rows);
     } catch (IOException e) {
-      LOG.error(e);
-      LOG.info(String.format("[QueryId:%d] sending failure response", request.getRequestId()));
+      LOGGER.error(e);
+      LOGGER.info(String.format("[QueryId:%d] sending failure response", request.getRequestId()));
       return createFailureResponse(request, e);
     }
   }
 
   ShutdownResponse handleShutdown(ShutdownRequest request) {
-    LOG.info("Shutting down worker...");
+    LOGGER.info("Shutting down worker...");
     SearchModeDetailQueryExecutor.shutdownThreadPool();
     SearchModeVectorDetailQueryExecutor.shutdownThreadPool();
-    LOG.info("Worker shut down");
+    LOGGER.info("Worker shut down");
     return new ShutdownResponse(Status.SUCCESS.ordinal(), "");
   }
 
@@ -85,13 +114,13 @@ class RequestHandler {
     CarbonTable table = CarbonTable.buildFromTableInfo(tableInfo);
     QueryModel queryModel = createQueryModel(table, request);
 
-    LOG.info(String.format("[QueryId:%d] %s, number of block: %d",
-        request.getRequestId(), queryModel.toString(), mbSplit.getAllSplits().size()));
+    LOGGER.info(String.format("[QueryId:%d] %s, number of block: %d", request.getRequestId(),
+        queryModel.toString(), mbSplit.getAllSplits().size()));
 
     // read all rows by the reader
     List<CarbonRow> rows = new LinkedList<>();
-    try (CarbonRecordReader<CarbonRow> reader =
-        new IndexedRecordReader(request.getRequestId(), table, queryModel)) {
+    try (CarbonRecordReader<CarbonRow> reader = new IndexedRecordReader(request.getRequestId(),
+        table, queryModel)) {
       reader.initialize(mbSplit, null);
 
       // loop to read required number of rows.
@@ -104,22 +133,18 @@ class RequestHandler {
     } catch (InterruptedException e) {
       throw new IOException(e);
     }
-    LOG.info(String.format("[QueryId:%d] scan completed, return %d rows",
+    LOGGER.info(String.format("[QueryId:%d] scan completed, return %d rows",
         request.getRequestId(), rows.size()));
     return rows;
   }
 
-
-
   private QueryModel createQueryModel(CarbonTable table, QueryRequest request) {
     String[] projectColumns = request.getProjectColumns();
     Expression filter = null;
     if (request.getFilterExpression() != null) {
       filter = request.getFilterExpression();
     }
-    return new QueryModelBuilder(table)
-        .projectColumns(projectColumns)
-        .filterExpression(filter)
+    return new QueryModelBuilder(table).projectColumns(projectColumns).filterExpression(filter)
         .build();
   }
 
@@ -144,4 +169,50 @@ class RequestHandler {
     return new QueryResponse(request.getRequestId(), Status.SUCCESS.ordinal(), "", output);
   }
 
+  public BaseResponse handleLoadData(LoadDataRequest request) {
+    DataLoadExecutor executor = null;
+    try {
+      CarbonLoadModel model = request.getModel();
+
+      JobID jobId = CarbonInputFormatUtil.getJobId(new Date(), 0);
+      CarbonInputFormatUtil.createJobTrackerID(new Date());
+      TaskID taskId = new TaskID(jobId, TaskType.MAP, 0);
+      TaskAttemptID taskAttemptId = new TaskAttemptID(taskId, 0);
+      Configuration configuration = new Configuration(hadoopConf);
+      StoreUtil.configureCSVInputFormat(configuration, model);
+      configuration.set(FileInputFormat.INPUT_DIR, model.getFactFilePath());
+      // Set up the attempt context required to use in the output committer.
+      TaskAttemptContext hadoopAttemptContext =
+          new TaskAttemptContextImpl(configuration, taskAttemptId);
+
+      CSVInputFormat format = new CSVInputFormat();
+      List<InputSplit> splits = format.getSplits(hadoopAttemptContext);
+
+      CarbonIterator<Object[]>[] readerIterators = new CSVRecordReaderIterator[splits.size()];
+      for (int index = 0; index < splits.size(); index++) {
+        readerIterators[index] = new CSVRecordReaderIterator(
+            format.createRecordReader(splits.get(index), hadoopAttemptContext), splits.get(index),
+            hadoopAttemptContext);
+      }
+
+      executor = new DataLoadExecutor();
+      executor.execute(model, conf.storeTempLocation(), readerIterators);
+
+      return new BaseResponse(Status.SUCCESS.ordinal(), "");
+    } catch (IOException e) {
+      LOGGER.error(e, "Failed to handle load data");
+      return new BaseResponse(Status.FAILURE.ordinal(), e.getMessage());
+    } catch (InterruptedException e) {
+      LOGGER.error(e, "Interrupted handle load data ");
+      return new BaseResponse(Status.FAILURE.ordinal(), e.getMessage());
+    } catch (Exception e) {
+      LOGGER.error(e, "Failed to execute load data ");
+      return new BaseResponse(Status.FAILURE.ordinal(), e.getMessage());
+    } finally {
+      if (executor != null) {
+        executor.close();
+        StoreUtil.clearUnsafeMemory(ThreadLocalTaskInfo.getCarbonTaskInfo().getTaskId());
+      }
+    }
+  }
 }


Mime
View raw message