You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2018/07/18 02:20:14 UTC
[41/50] [abbrv] carbondata git commit: [CARBONDATA-2690][CarbonStore]
implement RESTful API: create table, load data and select
[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/4437920a
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/4437920a
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/4437920a
Branch: refs/heads/carbonstore
Commit: 4437920a3174921d6397c4f596ff941e2cd0faa0
Parents: d9b40bf
Author: QiangCai <qi...@qq.com>
Authored: Tue Jul 3 20:21:18 2018 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Wed Jul 18 10:10:15 2018 +0800
----------------------------------------------------------------------
.../schema/table/CarbonTableBuilder.java | 9 +-
.../schema/table/TableSchemaBuilder.java | 27 +-
.../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 | 1 +
.../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, 5717 insertions(+), 865 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/4437920a/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/4437920a/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 4780e60..6f575a4 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,6 +54,8 @@ public class TableSchemaBuilder {
private List<ColumnSchema> measures = new LinkedList<>();
+ private Map<String, String> tableProperties;
+
private int blockSize;
private int blockletSize;
@@ -62,6 +64,14 @@ public class TableSchemaBuilder {
private boolean isLocalDictionaryEnabled;
private String localDictionaryThreshold;
+ 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");
@@ -111,22 +121,25 @@ 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));
}
// Adding local dictionary, applicable only for String(dictionary exclude)
if (isLocalDictionaryEnabled) {
- property.put(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE,
+ tableProperties.put(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE,
String.valueOf(isLocalDictionaryEnabled));
String localdictionaryThreshold = localDictionaryThreshold.equalsIgnoreCase("0") ?
CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD_DEFAULT :
localDictionaryThreshold;
- property.put(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD, localdictionaryThreshold);
+ tableProperties.put(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD, localdictionaryThreshold);
for (int index = 0; index < allColumns.size(); index++) {
ColumnSchema colSchema = allColumns.get(index);
if (colSchema.getDataType() == DataTypes.STRING
@@ -136,8 +149,8 @@ public class TableSchemaBuilder {
}
}
}
- 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/4437920a/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/4437920a/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/4437920a/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/4437920a/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 b49b6f9..f411722 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
@@ -65,7 +65,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/4437920a/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/4437920a/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/4437920a/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/4437920a/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 6c796b5..2e3d7aa 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonRecordReader.java
@@ -129,6 +129,7 @@ public class CarbonRecordReader<T> extends AbstractRecordReader<T> {
DataMapStoreManager.getInstance().clearDataMaps(
queryModel.getTable().getAbsoluteTableIdentifier());
}
+
// close read support
readSupport.close();
try {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index aeb87d3..b1cf347 100644
--- a/pom.xml
+++ b/pom.xml
@@ -585,6 +585,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/4437920a/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 0b60fa5..1ff7752 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
@@ -51,10 +51,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
@@ -65,7 +75,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++) {
@@ -74,10 +86,12 @@ public class CarbonLoadModelBuilder {
optionsFinal.put("fileheader", Strings.mkString(columns, ","));
}
optionsFinal.put("bad_record_path", CarbonBadRecordUtil.getBadRecordsPath(options, table));
+
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/4437920a/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 8a0c2b6..0ff71c7 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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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());
+ }
+ }
+ }
}