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 2016/12/29 14:35:31 UTC
[2/3] incubator-carbondata git commit: WIP provide dictionary
server/client framework
WIP provide dictionary server/client framework
complete the netty communication
netty
implement methods, and use new dictionarygenerator
trigger msg: initial, generate, size
add msg: write dictionary
fix write mutil dictionary file
ObjectDecoder
fixerror
fix query error
remove no used files
shutdown dictionary server and client
fixDictServerCloseBug and format code
testOnePass
fix second load
fix complex
fix multil level complex
optimize DictionaryKey
fix testcase
fix testcase and get dict cache
fix checkstyle
remove some useless judgement conditions
fix complex test case
fix testcase
fix mutil task
fix print
use json for serialize, and fix write dictionary issues
fix duplicate dictionary
write dictionary once
multi thread write dictionary
add testcase
RemoveUselessCode
fix comments
fix nullpointerexception
optimize
support predef column dictionary
use kryo
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/05b26549
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/05b26549
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/05b26549
Branch: refs/heads/master
Commit: 05b26549e929c2a461b0d18e5b58d3cd61bec1d5
Parents: 241f45f
Author: ravipesala <ra...@gmail.com>
Authored: Tue Nov 15 19:15:17 2016 +0530
Committer: jackylk <ja...@huawei.com>
Committed: Thu Dec 29 20:49:05 2016 +0800
----------------------------------------------------------------------
.../AbstractColumnDictionaryInfo.java | 2 +-
.../dictionary/AbstractDictionaryCache.java | 25 --
.../cache/dictionary/ColumnDictionaryInfo.java | 2 +-
.../dictionary/ColumnReverseDictionaryInfo.java | 2 +-
.../cache/dictionary/ForwardDictionary.java | 4 +-
.../cache/dictionary/ReverseDictionary.java | 4 +-
.../metadata/schema/table/CarbonTable.java | 62 +++--
.../core/constants/CarbonCommonConstants.java | 10 +
.../dictionary/client/DictionaryClient.java | 92 +++++++
.../client/DictionaryClientHandler.java | 109 +++++++++
.../dictionary/generator/DictionaryWriter.java | 29 +++
.../IncrementalColumnDictionaryGenerator.java | 241 +++++++++++++++++++
.../generator/ServerDictionaryGenerator.java | 74 ++++++
.../generator/TableDictionaryGenerator.java | 116 +++++++++
.../dictionary/generator/key/DictionaryKey.java | 92 +++++++
.../dictionary/generator/key/KryoRegister.java | 68 ++++++
.../dictionary/server/DictionaryServer.java | 93 +++++++
.../server/DictionaryServerHandler.java | 108 +++++++++
.../apache/carbondata/core/util/CarbonUtil.java | 27 +++
.../spark/util/GlobalDictionaryUtil.scala | 4 +-
.../spark/sql/catalyst/CarbonDDLSqlParser.scala | 3 +-
.../spark/rdd/CarbonDataRDDFactory.scala | 15 ++
.../execution/command/carbonTableSchema.scala | 79 +++++-
.../test/resources/columndictionary/country.csv | 5 +
.../test/resources/columndictionary/name.csv | 10 +
.../spark/src/test/resources/dataIncrement.csv | 21 ++
.../complexType/TestCreateTableWithDouble.scala | 2 +
.../dataload/TestLoadDataWithSinglePass.scala | 114 +++++++++
.../filterexpr/FilterProcessorTestCase.scala | 2 +-
.../processing/datatypes/PrimitiveDataType.java | 41 +++-
.../processing/model/CarbonLoadModel.java | 45 ++++
.../newflow/CarbonDataLoadConfiguration.java | 42 ++++
.../newflow/DataLoadProcessBuilder.java | 11 +
.../impl/DictionaryFieldConverterImpl.java | 52 +++-
.../converter/impl/FieldEncoderFactory.java | 23 +-
.../converter/impl/RowConverterImpl.java | 45 +++-
.../DictionaryServerClientDictionary.java | 95 ++++++++
37 files changed, 1692 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractColumnDictionaryInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractColumnDictionaryInfo.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractColumnDictionaryInfo.java
index ad766d7..ab400f0 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractColumnDictionaryInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractColumnDictionaryInfo.java
@@ -282,7 +282,7 @@ public abstract class AbstractColumnDictionaryInfo implements DictionaryInfo {
* 2. Filter scenarios where from value surrogate key has to be found.
*
* @param value dictionary value
- * @return if found returns key else 0
+ * @return if found returns key else INVALID_SURROGATE_KEY
*/
@Override public int getSurrogateKey(String value) {
byte[] keyData = value.getBytes(Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
index e3e6532..5fbef5d 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
@@ -35,7 +35,6 @@ import org.apache.carbondata.core.reader.CarbonDictionaryMetadataReader;
import org.apache.carbondata.core.service.DictionaryService;
import org.apache.carbondata.core.service.PathService;
import org.apache.carbondata.core.util.CarbonProperties;
-import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.CarbonUtilException;
/**
@@ -85,30 +84,6 @@ public abstract class AbstractDictionaryCache<K extends DictionaryColumnUniqueId
}
/**
- * This method will check if dictionary and its metadata file exists for a given column
- *
- * @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
- * tableName and columnIdentifier
- * @return
- */
- protected boolean isFileExistsForGivenColumn(
- DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
- PathService pathService = CarbonCommonFactory.getPathService();
- CarbonTablePath carbonTablePath = pathService.getCarbonTablePath(carbonStorePath,
- dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier());
-
- String dictionaryFilePath =
- carbonTablePath.getDictionaryFilePath(dictionaryColumnUniqueIdentifier
- .getColumnIdentifier().getColumnId());
- String dictionaryMetadataFilePath =
- carbonTablePath.getDictionaryMetaFilePath(dictionaryColumnUniqueIdentifier
- .getColumnIdentifier().getColumnId());
- // check if both dictionary and its metadata file exists for a given column
- return CarbonUtil.isFileExists(dictionaryFilePath) && CarbonUtil
- .isFileExists(dictionaryMetadataFilePath);
- }
-
- /**
* This method will read dictionary metadata file and return the dictionary meta chunks
*
* @param dictionaryColumnUniqueIdentifier
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnDictionaryInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnDictionaryInfo.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnDictionaryInfo.java
index 1d2eb8b..cd2a88c 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnDictionaryInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnDictionaryInfo.java
@@ -211,7 +211,7 @@ public class ColumnDictionaryInfo extends AbstractColumnDictionaryInfo {
return surrogateKey; // key found
}
}
- return 0;
+ return CarbonCommonConstants.INVALID_SURROGATE_KEY;
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnReverseDictionaryInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnReverseDictionaryInfo.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnReverseDictionaryInfo.java
index 03a6d6a..ff4f10a 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnReverseDictionaryInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ColumnReverseDictionaryInfo.java
@@ -64,7 +64,7 @@ public class ColumnReverseDictionaryInfo extends AbstractColumnDictionaryInfo {
* 2. Filter scenarios where from value surrogate key has to be found.
*
* @param value dictionary value as byte array. It will be treated as key here
- * @return if found returns key else 0
+ * @return if found returns key else INVALID_SURROGATE_KEY
*/
@Override public int getSurrogateKey(byte[] value) {
DictionaryByteArrayWrapper dictionaryByteArrayWrapper =
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionary.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionary.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionary.java
index bddab56..7d460d0 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionary.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ForwardDictionary.java
@@ -50,7 +50,7 @@ public class ForwardDictionary implements Dictionary {
* 2. Filter scenarios where from value surrogate key has to be found.
*
* @param value dictionary value
- * @return if found returns key else 0
+ * @return if found returns key else INVALID_SURROGATE_KEY
*/
@Override public int getSurrogateKey(String value) {
return columnDictionaryInfo.getSurrogateKey(value);
@@ -64,7 +64,7 @@ public class ForwardDictionary implements Dictionary {
* 2. Filter scenarios where from value surrogate key has to be found.
*
* @param value dictionary value as byte array
- * @return if found returns key else 0
+ * @return if found returns key else INVALID_SURROGATE_KEY
*/
@Override public int getSurrogateKey(byte[] value) {
return columnDictionaryInfo.getSurrogateKey(value);
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ReverseDictionary.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ReverseDictionary.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ReverseDictionary.java
index 1f8a3ff..cce4156 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ReverseDictionary.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ReverseDictionary.java
@@ -44,7 +44,7 @@ public class ReverseDictionary implements Dictionary {
* 2. Filter scenarios where from value surrogate key has to be found.
*
* @param value dictionary value
- * @return if found returns key else 0
+ * @return if found returns key else INVALID_SURROGATE_KEY
*/
@Override public int getSurrogateKey(String value) {
return columnReverseDictionaryInfo.getSurrogateKey(value);
@@ -58,7 +58,7 @@ public class ReverseDictionary implements Dictionary {
* 2. Filter scenarios where from value surrogate key has to be found.
*
* @param value dictionary value as byte array
- * @return if found returns key else 0
+ * @return if found returns key else INVALID_SURROGATE_KEY
*/
@Override public int getSurrogateKey(byte[] value) {
return columnReverseDictionaryInfo.getSurrogateKey(value);
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/carbon/metadata/schema/table/CarbonTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/metadata/schema/table/CarbonTable.java b/core/src/main/java/org/apache/carbondata/core/carbon/metadata/schema/table/CarbonTable.java
index 7766616..f756ab1 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/metadata/schema/table/CarbonTable.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/metadata/schema/table/CarbonTable.java
@@ -68,6 +68,11 @@ public class CarbonTable implements Serializable {
private Map<String, List<CarbonColumn>> createOrderColumn;
/**
+ * TableName, Dimensions and children dimensions list
+ */
+ private Map<String, List<CarbonDimension>> tablePrimitiveDimensionsMap;
+
+ /**
* table measures list.
*/
private Map<String, List<CarbonMeasure>> tableMeasuresMap;
@@ -108,6 +113,7 @@ public class CarbonTable implements Serializable {
this.tableBucketMap = new HashMap<>();
this.aggregateTablesName = new ArrayList<String>();
this.createOrderColumn = new HashMap<String, List<CarbonColumn>>();
+ this.tablePrimitiveDimensionsMap = new HashMap<String, List<CarbonDimension>>();
}
/**
@@ -195,8 +201,10 @@ public class CarbonTable implements Serializable {
*/
private void fillDimensionsAndMeasuresForTables(TableSchema tableSchema) {
List<CarbonDimension> dimensions = new ArrayList<CarbonDimension>();
+ List<CarbonDimension> primitiveDimensions = new ArrayList<CarbonDimension>();
List<CarbonMeasure> measures = new ArrayList<CarbonMeasure>();
this.tableDimensionsMap.put(tableSchema.getTableName(), dimensions);
+ this.tablePrimitiveDimensionsMap.put(this.tableUniqueName, primitiveDimensions);
this.tableMeasuresMap.put(tableSchema.getTableName(), measures);
int dimensionOrdinal = 0;
int measureOrdinal = 0;
@@ -216,26 +224,32 @@ public class CarbonTable implements Serializable {
dimensions.add(complexDimension);
dimensionOrdinal =
readAllComplexTypeChildrens(dimensionOrdinal, columnSchema.getNumberOfChild(),
- listOfColumns, complexDimension);
+ listOfColumns, complexDimension, primitiveDimensions);
i = dimensionOrdinal - 1;
complexTypeOrdinal = assignComplexOrdinal(complexDimension, complexTypeOrdinal);
} else {
if (!columnSchema.getEncodingList().contains(Encoding.DICTIONARY)) {
- dimensions.add(new CarbonDimension(columnSchema, dimensionOrdinal++,
- columnSchema.getSchemaOrdinal(), -1, -1, -1));
+ CarbonDimension dimension =
+ new CarbonDimension(columnSchema, dimensionOrdinal++,
+ columnSchema.getSchemaOrdinal(), -1, -1, -1);
+ dimensions.add(dimension);
+ primitiveDimensions.add(dimension);
} else if (columnSchema.getEncodingList().contains(Encoding.DICTIONARY)
&& columnSchema.getColumnGroupId() == -1) {
- dimensions
- .add(new CarbonDimension(columnSchema, dimensionOrdinal++,
- columnSchema.getSchemaOrdinal(), keyOrdinal++, -1, -1));
+ CarbonDimension dimension =
+ new CarbonDimension(columnSchema, dimensionOrdinal++,
+ columnSchema.getSchemaOrdinal(), keyOrdinal++, -1, -1);
+ dimensions.add(dimension);
+ primitiveDimensions.add(dimension);
} else {
columnGroupOrdinal =
previousColumnGroupId == columnSchema.getColumnGroupId() ? ++columnGroupOrdinal : 0;
previousColumnGroupId = columnSchema.getColumnGroupId();
- dimensions.add(new CarbonDimension(columnSchema, dimensionOrdinal++,
- columnSchema.getSchemaOrdinal(), keyOrdinal++,
- columnGroupOrdinal, -1));
-
+ CarbonDimension dimension = new CarbonDimension(columnSchema, dimensionOrdinal++,
+ columnSchema.getSchemaOrdinal(), keyOrdinal++,
+ columnGroupOrdinal, -1);
+ dimensions.add(dimension);
+ primitiveDimensions.add(dimension);
}
}
} else {
@@ -256,7 +270,8 @@ public class CarbonTable implements Serializable {
* @return
*/
private int readAllComplexTypeChildrens(int dimensionOrdinal, int childCount,
- List<ColumnSchema> listOfColumns, CarbonDimension parentDimension) {
+ List<ColumnSchema> listOfColumns, CarbonDimension parentDimension,
+ List<CarbonDimension> primitiveDimensions) {
for (int i = 0; i < childCount; i++) {
ColumnSchema columnSchema = listOfColumns.get(dimensionOrdinal);
if (columnSchema.isDimensionColumn()) {
@@ -268,11 +283,13 @@ public class CarbonTable implements Serializable {
parentDimension.getListOfChildDimensions().add(complexDimension);
dimensionOrdinal =
readAllComplexTypeChildrens(dimensionOrdinal, columnSchema.getNumberOfChild(),
- listOfColumns, complexDimension);
+ listOfColumns, complexDimension, primitiveDimensions);
} else {
- parentDimension.getListOfChildDimensions()
- .add(new CarbonDimension(columnSchema, dimensionOrdinal++,
- columnSchema.getSchemaOrdinal(), -1, -1, -1));
+ CarbonDimension carbonDimension =
+ new CarbonDimension(columnSchema, dimensionOrdinal++,
+ columnSchema.getSchemaOrdinal(), -1, -1, -1);
+ parentDimension.getListOfChildDimensions().add(carbonDimension);
+ primitiveDimensions.add(carbonDimension);
}
}
}
@@ -518,4 +535,19 @@ public class CarbonTable implements Serializable {
this.blockSize = blockSize;
}
+ /**
+ * to get the normal dimension or the primitive dimension of the complex type
+ *
+ * @param tableName
+ * @return primitive dimension of a table
+ */
+ public CarbonDimension getPrimitiveDimensionByName(String tableName, String columnName) {
+ List<CarbonDimension> dimList = tablePrimitiveDimensionsMap.get(tableName);
+ for (CarbonDimension dim : dimList) {
+ if (dim.getColName().equalsIgnoreCase(columnName)) {
+ return dim;
+ }
+ }
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index 2d1ba9f..f766042 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -981,6 +981,16 @@ public final class CarbonCommonConstants {
public static final String ENABLE_VECTOR_READER_DEFAULT = "false";
+ /*
+ * carbon dictionary server port
+ */
+ public static final String DICTIONARY_SERVER_PORT = "carbon.dictionary.server.port";
+
+ /**
+ * Default carbon dictionary server port
+ */
+ public static final String DICTIONARY_SERVER_PORT_DEFAULT = "2030";
+
private CarbonCommonConstants() {
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClient.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClient.java b/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClient.java
new file mode 100644
index 0000000..92bc56a
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClient.java
@@ -0,0 +1,92 @@
+/*
+ * 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.core.dictionary.client;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.dictionary.generator.key.DictionaryKey;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.channel.*;
+import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
+import org.jboss.netty.handler.codec.serialization.ClassResolvers;
+import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
+import org.jboss.netty.handler.codec.serialization.ObjectEncoder;
+
+
+/**
+ * Dictionary client to connect to Dictionary server and generate dictionary values
+ */
+public class DictionaryClient {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(DictionaryClient.class.getName());
+
+ private DictionaryClientHandler dictionaryClientHandler = new DictionaryClientHandler();
+
+ private ClientBootstrap clientBootstrap;
+
+ /**
+ * start dictionary client
+ *
+ * @param address
+ * @param port
+ */
+ public void startClient(String address, int port) {
+ clientBootstrap = new ClientBootstrap();
+ ExecutorService boss = Executors.newCachedThreadPool();
+ ExecutorService worker = Executors.newCachedThreadPool();
+ clientBootstrap.setFactory(new NioClientSocketChannelFactory(boss, worker));
+ clientBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
+ @Override
+ public ChannelPipeline getPipeline() throws Exception {
+ ChannelPipeline pipeline = Channels.pipeline();
+ pipeline.addLast("ObjectEncoder", new ObjectEncoder());
+ pipeline.addLast("ObjectDecoder", new ObjectDecoder(ClassResolvers.cacheDisabled(
+ getClass().getClassLoader())));
+ pipeline.addLast("DictionaryClientHandler", dictionaryClientHandler);
+ return pipeline;
+ }
+ });
+ clientBootstrap.connect(new InetSocketAddress(address, port));
+ LOGGER.audit("Client Start!");
+ }
+
+ /**
+ * for client request
+ *
+ * @param key
+ * @return
+ */
+ public DictionaryKey getDictionary(DictionaryKey key) {
+ return dictionaryClientHandler.getDictionary(key);
+ }
+
+ /**
+ * shutdown dictionary client
+ */
+ public void shutDown() {
+ clientBootstrap.releaseExternalResources();
+ clientBootstrap.shutdown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClientHandler.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClientHandler.java b/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClientHandler.java
new file mode 100644
index 0000000..d5ca781
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/client/DictionaryClientHandler.java
@@ -0,0 +1,109 @@
+/*
+ * 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.core.dictionary.client;
+
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.dictionary.generator.key.DictionaryKey;
+import org.apache.carbondata.core.dictionary.generator.key.KryoRegister;
+
+import org.jboss.netty.channel.*;
+
+/**
+ * Client handler to get data.
+ */
+public class DictionaryClientHandler extends SimpleChannelHandler {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(DictionaryClientHandler.class.getName());
+
+ final Map<String, BlockingQueue<DictionaryKey>> dictKeyQueueMap = new ConcurrentHashMap<>();
+
+ private ChannelHandlerContext ctx;
+
+ private Object lock = new Object();
+
+ @Override
+ public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+ this.ctx = ctx;
+ LOGGER.audit("Connected " + ctx.getHandler());
+ super.channelConnected(ctx, e);
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
+ byte[] response = (byte[]) e.getMessage();
+ DictionaryKey key = KryoRegister.deserialize(response);
+ BlockingQueue<DictionaryKey> dictKeyQueue = dictKeyQueueMap.get(key.getThreadNo());
+ dictKeyQueue.offer(key);
+ super.messageReceived(ctx, e);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
+ LOGGER.error("exceptionCaught");
+ e.getCause().printStackTrace();
+ ctx.getChannel().close();
+ }
+
+ /**
+ * client send request to server
+ *
+ * @param key
+ * @return
+ */
+ public DictionaryKey getDictionary(DictionaryKey key) {
+ DictionaryKey dictionaryKey;
+ BlockingQueue<DictionaryKey> dictKeyQueue = null;
+ try {
+ synchronized (lock) {
+ dictKeyQueue = dictKeyQueueMap.get(key.getThreadNo());
+ if (dictKeyQueue == null) {
+ dictKeyQueue = new LinkedBlockingQueue<DictionaryKey>();
+ dictKeyQueueMap.put(key.getThreadNo(), dictKeyQueue);
+ }
+ }
+ byte[] serialize = KryoRegister.serialize(key);
+ ctx.getChannel().write(serialize);
+ } catch (Exception e) {
+ LOGGER.error("Error while send request to server " + e.getMessage());
+ ctx.getChannel().close();
+ }
+ boolean interrupted = false;
+ try {
+ for (; ; ) {
+ try {
+ dictionaryKey = dictKeyQueue.take();
+ return dictionaryKey;
+ } catch (InterruptedException ignore) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/generator/DictionaryWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/DictionaryWriter.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/DictionaryWriter.java
new file mode 100644
index 0000000..8c160dc
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/DictionaryWriter.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.core.dictionary.generator;
+
+
+/**
+ * Interface to write dictionary to file system
+ */
+public interface DictionaryWriter {
+
+ void writeDictionaryData(String tableUniqueName) throws Exception;
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
new file mode 100644
index 0000000..fe84ef5
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGenerator.java
@@ -0,0 +1,241 @@
+/*
+ * 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.core.dictionary.generator;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.common.factory.CarbonCommonFactory;
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.Cache;
+import org.apache.carbondata.core.cache.CacheProvider;
+import org.apache.carbondata.core.cache.CacheType;
+import org.apache.carbondata.core.cache.dictionary.Dictionary;
+import org.apache.carbondata.core.cache.dictionary.DictionaryColumnUniqueIdentifier;
+import org.apache.carbondata.core.carbon.CarbonTableIdentifier;
+import org.apache.carbondata.core.carbon.ColumnIdentifier;
+import org.apache.carbondata.core.carbon.metadata.CarbonMetadata;
+import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.devapi.BiDictionary;
+import org.apache.carbondata.core.devapi.DictionaryGenerationException;
+import org.apache.carbondata.core.devapi.DictionaryGenerator;
+import org.apache.carbondata.core.service.DictionaryService;
+import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.carbondata.core.writer.CarbonDictionaryWriter;
+import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortIndexWriter;
+import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortInfo;
+import org.apache.carbondata.core.writer.sortindex.CarbonDictionarySortInfoPreparator;
+
+
+/**
+ * This generator does not maintain the whole cache of dictionary. It just maintains the cache only
+ * for the loading session, so what ever the dictionary values it generates in the loading session
+ * it keeps in cache.
+ */
+public class IncrementalColumnDictionaryGenerator implements BiDictionary<Integer, String>,
+ DictionaryGenerator<Integer, String>, DictionaryWriter {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(IncrementalColumnDictionaryGenerator.class.getName());
+
+ private final Object lock = new Object();
+
+ private Map<String, Integer> incrementalCache = new ConcurrentHashMap<>();
+
+ private Map<Integer, String> reverseIncrementalCache = new ConcurrentHashMap<>();
+
+ private int currentDictionarySize;
+
+ private CarbonDimension dimension;
+
+ public IncrementalColumnDictionaryGenerator(CarbonDimension dimension, int maxValue) {
+ this.currentDictionarySize = maxValue;
+ this.dimension = dimension;
+ }
+
+ @Override public Integer getOrGenerateKey(String value) throws DictionaryGenerationException {
+ Integer dict = getKey(value);
+ if (dict == null) {
+ dict = generateKey(value);
+ }
+ return dict;
+ }
+
+ @Override public Integer getKey(String value) {
+ return incrementalCache.get(value);
+ }
+
+ @Override public String getValue(Integer key) {
+ return reverseIncrementalCache.get(key);
+ }
+
+ @Override public int size() {
+ return currentDictionarySize;
+ }
+
+ @Override public Integer generateKey(String value) throws DictionaryGenerationException {
+ synchronized (lock) {
+ Integer dict = incrementalCache.get(value);
+ if (dict == null) {
+ dict = ++currentDictionarySize;
+ incrementalCache.put(value, dict);
+ reverseIncrementalCache.put(dict, value);
+ }
+ return dict;
+ }
+ }
+
+ @Override public void writeDictionaryData(String tableUniqueName) throws Exception {
+ // initialize params
+ CarbonMetadata metadata = CarbonMetadata.getInstance();
+ CarbonTable carbonTable = metadata.getCarbonTable(tableUniqueName);
+ CarbonTableIdentifier tableIdentifier = carbonTable.getCarbonTableIdentifier();
+ ColumnIdentifier columnIdentifier = dimension.getColumnIdentifier();
+ String storePath = carbonTable.getStorePath();
+ DictionaryService dictionaryService = CarbonCommonFactory.getDictionaryService();
+ // create dictionary cache from dictionary File
+ DictionaryColumnUniqueIdentifier identifier =
+ new DictionaryColumnUniqueIdentifier(tableIdentifier, columnIdentifier,
+ columnIdentifier.getDataType());
+ Boolean isDictExists = CarbonUtil.isFileExistsForGivenColumn(storePath, identifier);
+ Dictionary dictionary = null;
+ long t1 = System.currentTimeMillis();
+ if (isDictExists) {
+ Cache<DictionaryColumnUniqueIdentifier, Dictionary> dictCache = CacheProvider.getInstance()
+ .createCache(CacheType.REVERSE_DICTIONARY, storePath);
+ dictionary = dictCache.get(identifier);
+ }
+ long dictCacheTime = System.currentTimeMillis() - t1;
+ long t2 = System.currentTimeMillis();
+ // write dictionary
+ CarbonDictionaryWriter dictionaryWriter = null;
+ dictionaryWriter = dictionaryService
+ .getDictionaryWriter(tableIdentifier, columnIdentifier, storePath);
+ List<String> distinctValues = writeDictionary(dictionaryWriter, isDictExists);
+ long dictWriteTime = System.currentTimeMillis() - t2;
+ long t3 = System.currentTimeMillis();
+ // write sort index
+ if (distinctValues.size() > 0) {
+ writeSortIndex(distinctValues, dictionary,
+ dictionaryService, tableIdentifier, columnIdentifier, storePath);
+ }
+ long sortIndexWriteTime = System.currentTimeMillis() - t3;
+ // update Meta Data
+ updateMetaData(dictionaryWriter);
+ LOGGER.audit("\n columnName: " + dimension.getColName() +
+ "\n columnId: " + dimension.getColumnId() +
+ "\n new distinct values count: " + distinctValues.size() +
+ "\n create dictionary cache: " + dictCacheTime +
+ "\n sort list, distinct and write: " + dictWriteTime +
+ "\n write sort info: " + sortIndexWriteTime);
+ }
+
+ /**
+ * write dictionary to file
+ *
+ * @param dictionaryWriter
+ * @param isDictExists
+ * @return
+ * @throws IOException
+ */
+ private List<String> writeDictionary(CarbonDictionaryWriter dictionaryWriter,
+ Boolean isDictExists) throws IOException {
+ List<String> distinctValues = new ArrayList<>();
+ try {
+ if (!isDictExists) {
+ dictionaryWriter.write(CarbonCommonConstants.MEMBER_DEFAULT_VAL);
+ distinctValues.add(CarbonCommonConstants.MEMBER_DEFAULT_VAL);
+ }
+ // write value to dictionary file
+ if (reverseIncrementalCache.size() > 0) {
+ for (int index = 2; index < reverseIncrementalCache.size() + 2; index++) {
+ String value = reverseIncrementalCache.get(index);
+ String parsedValue = DataTypeUtil
+ .normalizeColumnValueForItsDataType(value, dimension);
+ if (null != parsedValue) {
+ dictionaryWriter.write(parsedValue);
+ distinctValues.add(parsedValue);
+ }
+ }
+ }
+ } catch (IOException ex) {
+ throw ex;
+ } finally {
+ if (null != dictionaryWriter) {
+ dictionaryWriter.close();
+ }
+ }
+
+ return distinctValues;
+ }
+
+ /**
+ * write dictionary sort index to file
+ *
+ * @param distinctValues
+ * @param dictionary
+ * @param dictionaryService
+ * @param tableIdentifier
+ * @param columnIdentifier
+ * @param storePath
+ * @throws IOException
+ */
+ private void writeSortIndex(List<String> distinctValues,
+ Dictionary dictionary,
+ DictionaryService dictionaryService,
+ CarbonTableIdentifier tableIdentifier,
+ ColumnIdentifier columnIdentifier,
+ String storePath) throws Exception{
+ CarbonDictionarySortIndexWriter carbonDictionarySortIndexWriter = null;
+ try {
+ CarbonDictionarySortInfoPreparator preparator = new CarbonDictionarySortInfoPreparator();
+ CarbonDictionarySortInfo dictionarySortInfo =
+ preparator.getDictionarySortInfo(distinctValues, dictionary,
+ dimension.getDataType());
+ carbonDictionarySortIndexWriter =
+ dictionaryService.getDictionarySortIndexWriter(tableIdentifier, columnIdentifier,
+ storePath);
+ carbonDictionarySortIndexWriter.writeSortIndex(dictionarySortInfo.getSortIndex());
+ carbonDictionarySortIndexWriter
+ .writeInvertedSortIndex(dictionarySortInfo.getSortIndexInverted());
+ } finally {
+ if (null != carbonDictionarySortIndexWriter) {
+ carbonDictionarySortIndexWriter.close();
+ }
+ }
+ }
+
+ /**
+ * update dictionary metadata
+ *
+ * @param dictionaryWriter
+ * @throws IOException
+ */
+ private void updateMetaData(CarbonDictionaryWriter dictionaryWriter) throws IOException{
+ if (null != dictionaryWriter) {
+ dictionaryWriter.commit();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGenerator.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGenerator.java
new file mode 100644
index 0000000..594ec41
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGenerator.java
@@ -0,0 +1,74 @@
+/*
+ * 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.core.dictionary.generator;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.core.carbon.metadata.CarbonMetadata;
+import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension;
+import org.apache.carbondata.core.devapi.DictionaryGenerationException;
+import org.apache.carbondata.core.devapi.DictionaryGenerator;
+import org.apache.carbondata.core.dictionary.generator.key.DictionaryKey;
+
+/**
+ * This is the dictionary generator for all tables. It generates dictionary
+ * based on @{@link DictionaryKey}.
+ */
+public class ServerDictionaryGenerator implements DictionaryGenerator<Integer, DictionaryKey> {
+
+ /**
+ * the map of tableName to TableDictionaryGenerator
+ */
+ private Map<String, TableDictionaryGenerator> tableMap = new ConcurrentHashMap<>();
+
+ @Override public Integer generateKey(DictionaryKey value) throws DictionaryGenerationException {
+ TableDictionaryGenerator generator = tableMap.get(value.getTableUniqueName());
+ assert generator != null : "Table initialization for generator is not done";
+ return generator.generateKey(value);
+ }
+
+ public void initializeGeneratorForTable(DictionaryKey key) {
+ CarbonMetadata metadata = CarbonMetadata.getInstance();
+ CarbonTable carbonTable = metadata.getCarbonTable(key.getTableUniqueName());
+ CarbonDimension dimension = carbonTable.getPrimitiveDimensionByName(
+ key.getTableUniqueName(), key.getColumnName());
+ // initialize TableDictionaryGenerator first
+ if (tableMap.get(key.getTableUniqueName()) == null) {
+ tableMap.put(key.getTableUniqueName(), new TableDictionaryGenerator(dimension));
+ } else {
+ tableMap.get(key.getTableUniqueName()).updateGenerator(dimension);
+ }
+ }
+
+ public Integer size(DictionaryKey key) {
+ TableDictionaryGenerator generator = tableMap.get(key.getTableUniqueName());
+ assert generator != null : "Table intialization for generator is not done";
+ return generator.size(key);
+ }
+
+ public void writeDictionaryData() throws Exception {
+ for (String tableUniqueName: tableMap.keySet()) {
+ TableDictionaryGenerator generator = tableMap.get(tableUniqueName);
+ ((DictionaryWriter) generator).writeDictionaryData(tableUniqueName);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGenerator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGenerator.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGenerator.java
new file mode 100644
index 0000000..addc3b6
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGenerator.java
@@ -0,0 +1,116 @@
+/*
+ * 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.core.dictionary.generator;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.carbon.metadata.CarbonMetadata;
+import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.devapi.BiDictionary;
+import org.apache.carbondata.core.devapi.DictionaryGenerationException;
+import org.apache.carbondata.core.devapi.DictionaryGenerator;
+import org.apache.carbondata.core.dictionary.generator.key.DictionaryKey;
+import org.apache.carbondata.core.util.CarbonProperties;
+
+/**
+ * Dictionary generation for table.
+ */
+public class TableDictionaryGenerator
+ implements DictionaryGenerator<Integer, DictionaryKey>, DictionaryWriter {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(TableDictionaryGenerator.class.getName());
+
+ /**
+ * the map of columnName to dictionaryGenerator
+ */
+ private Map<String, DictionaryGenerator<Integer, String>> columnMap = new ConcurrentHashMap<>();
+
+ public TableDictionaryGenerator(CarbonDimension dimension) {
+ columnMap.put(dimension.getColumnId(),
+ new IncrementalColumnDictionaryGenerator(dimension, 1));
+ }
+
+ @Override public Integer generateKey(DictionaryKey value) throws DictionaryGenerationException {
+ CarbonMetadata metadata = CarbonMetadata.getInstance();
+ CarbonTable carbonTable = metadata.getCarbonTable(value.getTableUniqueName());
+ CarbonDimension dimension = carbonTable.getPrimitiveDimensionByName(
+ value.getTableUniqueName(), value.getColumnName());
+
+ DictionaryGenerator<Integer, String> generator =
+ columnMap.get(dimension.getColumnId());
+ return generator.generateKey(value.getData().toString());
+ }
+
+ public Integer size(DictionaryKey key) {
+ CarbonMetadata metadata = CarbonMetadata.getInstance();
+ CarbonTable carbonTable = metadata.getCarbonTable(key.getTableUniqueName());
+ CarbonDimension dimension = carbonTable.getPrimitiveDimensionByName(
+ key.getTableUniqueName(), key.getColumnName());
+
+ DictionaryGenerator<Integer, String> generator =
+ columnMap.get(dimension.getColumnId());
+ return ((BiDictionary) generator).size();
+ }
+
+ @Override public void writeDictionaryData(String tableUniqueName) throws Exception {
+ int numOfCores = 1;
+ final String tableName = tableUniqueName;
+ try {
+ numOfCores = Integer.parseInt(CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.NUM_CORES_LOADING,
+ CarbonCommonConstants.NUM_CORES_DEFAULT_VAL));
+ } catch (NumberFormatException e) {
+ numOfCores = Integer.parseInt(CarbonCommonConstants.NUM_CORES_DEFAULT_VAL);
+ }
+ long start = System.currentTimeMillis();
+ List<Future<Void>> taskSubmitList =
+ new ArrayList<>(columnMap.size());
+ ExecutorService executorService = Executors.newFixedThreadPool(numOfCores);
+ for (final DictionaryGenerator generator: columnMap.values()) {
+ taskSubmitList.add(executorService.submit(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ ((DictionaryWriter) (generator)).writeDictionaryData(tableName);
+ return null;
+ }
+ }));
+ }
+
+ try {
+ executorService.shutdown();
+ executorService.awaitTermination(1, TimeUnit.HOURS);
+ } catch (InterruptedException e) {
+ LOGGER.error("Error loading the dictionary: " + e.getMessage());
+ }
+ LOGGER.audit("Total time taken to write dictionary file is: " +
+ (System.currentTimeMillis() - start));
+ }
+
+ public void updateGenerator(CarbonDimension dimension) {
+ columnMap.put(dimension.getColumnId(),
+ new IncrementalColumnDictionaryGenerator(dimension, 1));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/DictionaryKey.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/DictionaryKey.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/DictionaryKey.java
new file mode 100644
index 0000000..dc4bfc3
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/DictionaryKey.java
@@ -0,0 +1,92 @@
+/*
+ * 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.core.dictionary.generator.key;
+
+import java.io.Serializable;
+
+/**
+ * Dictionary key to generate dictionary
+ */
+public class DictionaryKey implements Serializable {
+
+ /**
+ * tableUniqueName
+ */
+ private String tableUniqueName;
+
+ /**
+ * columnName
+ */
+ private String columnName;
+
+ /**
+ * message data
+ */
+ private Object data;
+
+ /**
+ * message type
+ */
+ private String type;
+
+ /**
+ * dictionary client thread no
+ */
+ private String threadNo;
+
+ public String getTableUniqueName() {
+ return tableUniqueName;
+ }
+
+ public String getColumnName() {
+ return columnName;
+ }
+
+ public Object getData() {
+ return data;
+ }
+
+ public void setData(Object data) {
+ this.data = data;
+ }
+
+ public void setThreadNo(String threadNo) {
+ this.threadNo = threadNo;
+ }
+
+ public String getThreadNo() {
+ return this.threadNo;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ public void setTableUniqueName(String tableUniqueName) {
+ this.tableUniqueName = tableUniqueName;
+ }
+
+ public void setColumnName(String columnName) {
+ this.columnName = columnName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/KryoRegister.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/KryoRegister.java b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/KryoRegister.java
new file mode 100644
index 0000000..d66b79a
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/generator/key/KryoRegister.java
@@ -0,0 +1,68 @@
+/*
+ * 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.core.dictionary.generator.key;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public class KryoRegister {
+ /**
+ * deserialize byte to DictionaryKey
+ *
+ * @param bb
+ * @return
+ */
+ public static DictionaryKey deserialize(byte[] bb) {
+ Kryo kryo = new Kryo();
+ kryo.setReferences(false);
+ kryo.setRegistrationRequired(true);
+ // register
+ Registration registration = kryo.register(DictionaryKey.class);
+
+ // deserialize
+ Input input = null;
+ input = new Input(bb);
+ DictionaryKey key = (DictionaryKey) kryo.readObject(input, registration.getType());
+ input.close();
+ return key;
+ }
+
+ /**
+ * serialize DictionaryKey to byte
+ *
+ * @param key
+ * @return
+ */
+ public static byte[] serialize(DictionaryKey key) {
+ Kryo kryo = new Kryo();
+ kryo.setReferences(false);
+ kryo.setRegistrationRequired(true);
+ // register
+ Registration registration = kryo.register(DictionaryKey.class);
+ //serialize
+ Output output = null;
+ output = new Output(1, 4096);
+ kryo.writeObject(output, key);
+ byte[] bb = output.toBytes();
+ output.flush();
+ return bb;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServer.java b/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServer.java
new file mode 100644
index 0000000..bb0e8f8
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServer.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.carbondata.core.dictionary.server;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.dictionary.generator.key.DictionaryKey;
+
+import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
+import org.jboss.netty.handler.codec.serialization.ClassResolvers;
+import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
+import org.jboss.netty.handler.codec.serialization.ObjectEncoder;
+
+
+/**
+ * Dictionary Server to generate dictionary keys.
+ */
+public class DictionaryServer {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(DictionaryServer.class.getName());
+
+ private ServerBootstrap bootstrap;
+
+ private DictionaryServerHandler dictionaryServerHandler;
+
+ /**
+ * start dictionary server
+ *
+ * @param port
+ * @throws Exception
+ */
+ public void startServer(int port) throws Exception {
+ bootstrap = new ServerBootstrap();
+ dictionaryServerHandler = new DictionaryServerHandler();
+
+ ExecutorService boss = Executors.newCachedThreadPool();
+ ExecutorService worker = Executors.newCachedThreadPool();
+
+ bootstrap.setFactory(new NioServerSocketChannelFactory(boss, worker));
+
+ bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
+ @Override
+ public ChannelPipeline getPipeline() throws Exception {
+ ChannelPipeline pipeline = Channels.pipeline();
+ pipeline.addLast("ObjectDecoder", new ObjectDecoder(ClassResolvers.cacheDisabled(
+ getClass().getClassLoader())));
+ pipeline.addLast("ObjectEncoder", new ObjectEncoder());
+ pipeline.addLast("DictionaryServerHandler", dictionaryServerHandler);
+ return pipeline;
+ }
+ });
+ bootstrap.bind(new InetSocketAddress(port));
+ LOGGER.audit("Server Start!");
+ }
+
+ /**
+ * shutdown dictionary server
+ *
+ * @throws Exception
+ */
+ public void shutdown() throws Exception {
+ DictionaryKey key = new DictionaryKey();
+ key.setType("WRITE_DICTIONARY");
+ dictionaryServerHandler.processMessage(key);
+ bootstrap.releaseExternalResources();
+ bootstrap.shutdown();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServerHandler.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServerHandler.java b/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServerHandler.java
new file mode 100644
index 0000000..9160ffe
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/dictionary/server/DictionaryServerHandler.java
@@ -0,0 +1,108 @@
+/*
+ * 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.core.dictionary.server;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.dictionary.generator.ServerDictionaryGenerator;
+import org.apache.carbondata.core.dictionary.generator.key.DictionaryKey;
+import org.apache.carbondata.core.dictionary.generator.key.KryoRegister;
+
+import org.jboss.netty.channel.*;
+
+/**
+ * Handler for Dictionary server.
+ */
+public class DictionaryServerHandler extends SimpleChannelHandler {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(DictionaryServerHandler.class.getName());
+
+ /**
+ * dictionary generator
+ */
+ private ServerDictionaryGenerator generatorForServer = new ServerDictionaryGenerator();
+
+ /**
+ * channel connected
+ *
+ * @param ctx
+ * @param e
+ * @throws Exception
+ */
+ public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+ LOGGER.audit("Connected " + ctx.getHandler());
+ }
+
+ /**
+ * receive message and handle
+ *
+ * @param ctx
+ * @param e
+ * @throws Exception
+ */
+ @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
+ throws Exception {
+ byte[] request = (byte[]) e.getMessage();
+ DictionaryKey key = KryoRegister.deserialize(request);
+ int outPut = processMessage(key);
+ key.setData(outPut);
+ // Send back the response
+ byte[] response = KryoRegister.serialize(key);
+ ctx.getChannel().write(response);
+ super.messageReceived(ctx, e);
+ }
+
+ /**
+ * handle exceptions
+ *
+ * @param ctx
+ * @param e
+ */
+ @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
+ LOGGER.error("exceptionCaught");
+ e.getCause().printStackTrace();
+ ctx.getChannel().close();
+ }
+
+ /**
+ * process message by message type
+ *
+ * @param key
+ * @return
+ * @throws Exception
+ */
+ public Integer processMessage(DictionaryKey key) throws Exception {
+ switch (key.getType()) {
+ case "DICTIONARY_GENERATION":
+ return generatorForServer.generateKey(key);
+ case "TABLE_INTIALIZATION":
+ generatorForServer.initializeGeneratorForTable(key);
+ return 0;
+ case "SIZE":
+ return generatorForServer.size(key);
+ case "WRITE_DICTIONARY":
+ generatorForServer.writeDictionaryData();
+ return 0;
+ default:
+ return -1;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index fbbed76..78cfc36 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -42,9 +42,11 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
+import org.apache.carbondata.common.factory.CarbonCommonFactory;
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.cache.dictionary.DictionaryColumnUniqueIdentifier;
import org.apache.carbondata.core.carbon.AbsoluteTableIdentifier;
import org.apache.carbondata.core.carbon.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.carbon.datastore.chunk.impl.FixedLengthDimensionDataChunk;
@@ -67,6 +69,7 @@ import org.apache.carbondata.core.keygenerator.mdkey.NumberCompressor;
import org.apache.carbondata.core.metadata.ValueEncoderMeta;
import org.apache.carbondata.core.reader.ThriftReader;
import org.apache.carbondata.core.reader.ThriftReader.TBaseCreator;
+import org.apache.carbondata.core.service.PathService;
import org.apache.carbondata.format.DataChunk2;
import org.apache.carbondata.scan.model.QueryDimension;
@@ -1355,5 +1358,29 @@ public final class CarbonUtil {
}
return outputArray;
}
+
+
+ /**
+ * This method will check if dictionary and its metadata file exists for a given column
+ *
+ * @param dictionaryColumnUniqueIdentifier unique identifier which contains dbName,
+ * tableName and columnIdentifier
+ * @return
+ */
+ public static boolean isFileExistsForGivenColumn(String carbonStorePath,
+ DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) {
+ PathService pathService = CarbonCommonFactory.getPathService();
+ CarbonTablePath carbonTablePath = pathService.getCarbonTablePath(carbonStorePath,
+ dictionaryColumnUniqueIdentifier.getCarbonTableIdentifier());
+
+ String dictionaryFilePath =
+ carbonTablePath.getDictionaryFilePath(dictionaryColumnUniqueIdentifier
+ .getColumnIdentifier().getColumnId());
+ String dictionaryMetadataFilePath =
+ carbonTablePath.getDictionaryMetaFilePath(dictionaryColumnUniqueIdentifier
+ .getColumnIdentifier().getColumnId());
+ // check if both dictionary and its metadata file exists for a given column
+ return isFileExists(dictionaryFilePath) && isFileExists(dictionaryMetadataFilePath);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
index 3274e58..5d6badd 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
@@ -518,13 +518,13 @@ object GlobalDictionaryUtil {
* @param hdfsLocation store location on hdfs
* @param dictFolderPath generated global dict file path
*/
- private def generatePredefinedColDictionary(colDictFilePath: String,
+ def generatePredefinedColDictionary(colDictFilePath: String,
table: CarbonTableIdentifier,
dimensions: Array[CarbonDimension],
carbonLoadModel: CarbonLoadModel,
sqlContext: SQLContext,
hdfsLocation: String,
- dictFolderPath: String) = {
+ dictFolderPath: String): Unit = {
// set pre defined dictionary column
setPredefinedColumnDictPath(carbonLoadModel, colDictFilePath, table, dimensions)
val dictLoadModel = createDictionaryLoadModel(carbonLoadModel, table, dimensions,
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
index 461633d..ee96c35 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
@@ -720,7 +720,8 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser {
val supportedOptions = Seq("DELIMITER", "QUOTECHAR", "FILEHEADER", "ESCAPECHAR", "MULTILINE",
"COMPLEX_DELIMITER_LEVEL_1", "COMPLEX_DELIMITER_LEVEL_2", "COLUMNDICT",
"SERIALIZATION_NULL_FORMAT", "BAD_RECORDS_LOGGER_ENABLE", "BAD_RECORDS_ACTION",
- "ALL_DICTIONARY_PATH", "MAXCOLUMNS", "COMMENTCHAR", "USE_KETTLE", "DATEFORMAT"
+ "ALL_DICTIONARY_PATH", "MAXCOLUMNS", "COMMENTCHAR", "USE_KETTLE", "DATEFORMAT",
+ "SINGLE_PASS"
)
var isSupported = true
val invalidOptions = StringBuilder.newBuilder
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index d975502..1bfd0c7 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -43,6 +43,7 @@ import org.apache.carbondata.core.carbon.datastore.block.{Distributable, TableBl
import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.dictionary.server.DictionaryServer
import org.apache.carbondata.core.load.{BlockDetails, LoadMetadataDetails}
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.lcm.locks.{CarbonLockFactory, CarbonLockUtil, ICarbonLock, LockUsage}
@@ -341,6 +342,7 @@ object CarbonDataRDDFactory {
columnar: Boolean,
partitionStatus: String = CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS,
useKettle: Boolean,
+ result: Future[DictionaryServer],
dataFrame: Option[DataFrame] = None): Unit = {
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
val isAgg = false
@@ -759,6 +761,19 @@ object CarbonDataRDDFactory {
// TODO : Handle it
LOGGER.info("********Database updated**********")
}
+
+ // write dictionary file and shutdown dictionary server
+ if (carbonLoadModel.getUseOnePass) {
+ try {
+ result.get().shutdown()
+ } catch {
+ case ex: Exception =>
+ LOGGER.error("Error while close dictionary server and write dictionary file for " +
+ s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ throw new Exception("Dataload failed due to error while write dictionary file!")
+ }
+ }
+
LOGGER.audit("Data load is successful for " +
s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
try {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index d13346a..30978e4 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -18,10 +18,15 @@
package org.apache.spark.sql.execution.command
import java.io.File
+import java.util.concurrent.Callable
+import java.util.concurrent.Executors
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Future
import scala.collection.JavaConverters._
import scala.language.implicitConversions
+import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Literal}
@@ -41,6 +46,7 @@ import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDime
import org.apache.carbondata.core.carbon.path.CarbonStorePath
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastorage.store.impl.FileFactory
+import org.apache.carbondata.core.dictionary.server.DictionaryServer
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
import org.apache.carbondata.lcm.locks.{CarbonLockFactory, LockUsage}
import org.apache.carbondata.lcm.status.SegmentStatusManager
@@ -371,6 +377,24 @@ case class LoadTable(
carbonLoadModel
.setBadRecordsAction(
TableOptionConstant.BAD_RECORDS_ACTION.getName + "," + badRecordsLoggerRedirect)
+ // when single_pass=true, and use_kettle=false, and not use all dict
+ val useOnePass = options.getOrElse("single_pass", "false").trim.toLowerCase match {
+ case "true" =>
+ if (!useKettle && StringUtils.isEmpty(allDictionaryPath)) {
+ true
+ } else {
+ LOGGER.error("Can't use single_pass, because SINGLE_PASS and ALL_DICTIONARY_PATH" +
+ "can not be used together, and USE_KETTLE must be set as false")
+ false
+ }
+ case "false" =>
+ false
+ case illegal =>
+ LOGGER.error(s"Can't use single_pass, because illegal syntax found: [" + illegal + "] " +
+ "Please set it as 'true' or 'false'")
+ false
+ }
+ carbonLoadModel.setUseOnePass(useOnePass)
if (delimiter.equalsIgnoreCase(complex_delimiter_level_1) ||
complex_delimiter_level_1.equalsIgnoreCase(complex_delimiter_level_2) ||
@@ -387,6 +411,10 @@ case class LoadTable(
carbonLoadModel.setAllDictPath(allDictionaryPath)
val partitionStatus = CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS
+
+ var result: Future[DictionaryServer] = null
+ var executorService: ExecutorService = null
+
try {
// First system has to partition the data first and then call the load data
LOGGER.info(s"Initiating Direct Load for the Table : ($dbName.$tableName)")
@@ -396,9 +424,49 @@ case class LoadTable(
carbonLoadModel.setColDictFilePath(columnDict)
carbonLoadModel.setDirectLoad(true)
GlobalDictionaryUtil.updateTableMetadataFunc = updateTableMetadata
- GlobalDictionaryUtil
- .generateGlobalDictionary(sqlContext, carbonLoadModel, relation.tableMeta.storePath,
- dataFrame)
+
+ if (carbonLoadModel.getUseOnePass) {
+ val colDictFilePath = carbonLoadModel.getColDictFilePath
+ if (colDictFilePath != null) {
+ val storePath = relation.tableMeta.storePath
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val carbonTableIdentifier = carbonTable.getAbsoluteTableIdentifier
+ .getCarbonTableIdentifier
+ val carbonTablePath = CarbonStorePath
+ .getCarbonTablePath(storePath, carbonTableIdentifier)
+ val dictFolderPath = carbonTablePath.getMetadataDirectoryPath
+ val dimensions = carbonTable.getDimensionByTableName(
+ carbonTable.getFactTableName).asScala.toArray
+ carbonLoadModel.initPredefDictMap()
+ // generate predefined dictionary
+ GlobalDictionaryUtil
+ .generatePredefinedColDictionary(colDictFilePath, carbonTableIdentifier,
+ dimensions, carbonLoadModel, sqlContext, storePath, dictFolderPath)
+ }
+ // dictionaryServerClient dictionary generator
+ val dictionaryServerPort = CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.DICTIONARY_SERVER_PORT,
+ CarbonCommonConstants.DICTIONARY_SERVER_PORT_DEFAULT)
+ carbonLoadModel.setDictionaryServerPort(Integer.parseInt(dictionaryServerPort))
+ val sparkDriverHost = sqlContext.sparkContext.getConf.get("spark.driver.host")
+ carbonLoadModel.setDictionaryServerHost(sparkDriverHost)
+ // start dictionary server when use one pass load.
+ executorService = Executors.newFixedThreadPool(1)
+ result = executorService.submit(new Callable[DictionaryServer]() {
+ @throws[Exception]
+ def call: DictionaryServer = {
+ Thread.currentThread().setName("Dictionary server")
+ val server: DictionaryServer = new DictionaryServer
+ server.startServer(dictionaryServerPort.toInt)
+ server
+ }
+ })
+ } else {
+ GlobalDictionaryUtil
+ .generateGlobalDictionary(sqlContext, carbonLoadModel, relation.tableMeta.storePath,
+ dataFrame)
+ }
+
CarbonDataRDDFactory.loadCarbonData(sqlContext,
carbonLoadModel,
relation.tableMeta.storePath,
@@ -406,6 +474,7 @@ case class LoadTable(
columinar,
partitionStatus,
useKettle,
+ result,
dataFrame)
} catch {
case ex: Exception =>
@@ -415,6 +484,10 @@ case class LoadTable(
} finally {
// Once the data load is successful delete the unwanted partition files
try {
+ // shutdown dictionary server thread
+ if (carbonLoadModel.getUseOnePass) {
+ executorService.shutdownNow()
+ }
val fileType = FileFactory.getFileType(partitionLocation)
if (FileFactory.isFileExist(partitionLocation, fileType)) {
val file = FileFactory
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/test/resources/columndictionary/country.csv
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/resources/columndictionary/country.csv b/integration/spark/src/test/resources/columndictionary/country.csv
new file mode 100644
index 0000000..1fd8032
--- /dev/null
+++ b/integration/spark/src/test/resources/columndictionary/country.csv
@@ -0,0 +1,5 @@
+usa
+china
+uk
+france
+brazil
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/test/resources/columndictionary/name.csv
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/resources/columndictionary/name.csv b/integration/spark/src/test/resources/columndictionary/name.csv
new file mode 100644
index 0000000..cbd6eef
--- /dev/null
+++ b/integration/spark/src/test/resources/columndictionary/name.csv
@@ -0,0 +1,10 @@
+aaa1
+aaa2
+aaa3
+aaa4
+aaa5
+aaa6
+aaa7
+aaa8
+aaa9
+aaa10
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/test/resources/dataIncrement.csv
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/resources/dataIncrement.csv b/integration/spark/src/test/resources/dataIncrement.csv
new file mode 100644
index 0000000..04a84cc
--- /dev/null
+++ b/integration/spark/src/test/resources/dataIncrement.csv
@@ -0,0 +1,21 @@
+ID,date,country,name,phonetype,serialname,salary
+1,2015/7/23,china,aaa1,phone197,ASD69643,15000
+2,2015/7/24,china,aaa2,phone756,ASD42892,15001
+3,2015/7/25,china,aaa3,phone1904,ASD37014,15002
+4,2015/7/26,china,aaa4,phone2435,ASD66902,15003
+5,2015/7/27,china,aaa5,phone2441,ASD90633,15004
+6,2015/7/28,china,aaa6,phone294,ASD59961,15005
+7,2015/7/29,china,aaa7,phone610,ASD14875,15006
+8,2015/7/30,china,aaa8,phone1848,ASD57308,15007
+9,2015/7/18,china,aaa9,phone706,ASD86717,15008
+10,2015/7/19,usa,aaa10,phone685,ASD30505,15009
+1001,2015/7/8,china,aaa1001,phone13181,ASD850271,16001
+1002,2015/9/8,china,aaa1002,phone13182,ASD850272,16002
+1003,2015/8/8,usa,aaa1003,phone13183,ASD850273,16003
+1004,2015/8/8,china,aaa1004,phone13184,ASD850274,16004
+1005,2015/7/8,india,aaa1005,phone13185,ASD850275,16005
+1006,2015/7/8,china,aaa1006,phone13186,ASD850276,16006
+1007,2015/7/8,england,aaa1007,phone13187,ASD850277,16007
+1008,2015/7/8,china,aaa1008,phone13188,ASD850278,16008
+1009,2015/7/8,japan,aaa1009,phone13189,ASD850279,16009
+1010,2015/7/8,china,aaa1010,phone131810,ASD8502710,16010
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestCreateTableWithDouble.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestCreateTableWithDouble.scala b/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestCreateTableWithDouble.scala
index cb10baf..b9c8a68 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestCreateTableWithDouble.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestCreateTableWithDouble.scala
@@ -44,6 +44,8 @@ class TestCreateTableWithDouble extends QueryTest with BeforeAndAfterAll {
.getCanonicalPath + "/src/test/resources/sampleComplex.csv"
countNum = Array(Row(0))
doubleField = Array(Row(0))
+ sql("drop table if exists doubleComplex")
+ sql("drop table if exists doubleComplex2")
}
test("test creating carbon table with double in complex type") {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataWithSinglePass.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataWithSinglePass.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataWithSinglePass.scala
new file mode 100644
index 0000000..3e0b4c8
--- /dev/null
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataWithSinglePass.scala
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.carbondata.spark.testsuite.dataload
+
+import org.apache.spark.sql.common.util.CarbonHiveContext._
+import org.apache.spark.sql.common.util.QueryTest
+import org.scalatest.BeforeAndAfterAll
+
+/**
+ * Test Class for data loading use one pass
+ *
+ */
+class TestLoadDataWithSinglePass extends QueryTest with BeforeAndAfterAll {
+
+ override def beforeAll {
+ sql("DROP TABLE IF EXISTS table_two_pass")
+ sql("DROP TABLE IF EXISTS table_one_pass")
+ sql("DROP TABLE IF EXISTS table_one_pass_2")
+
+ sql(
+ """
+ |CREATE TABLE table_two_pass (ID int, date Timestamp, country String,
+ |name String, phonetype String, serialname String, salary int)
+ |STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+
+ sql(
+ """
+ |LOAD DATA local inpath './src/test/resources/dataDiff.csv' INTO TABLE table_two_pass
+ |OPTIONS('DELIMITER'= ',', 'USE_KETTLE'='false', 'SINGLE_PASS'='false')
+ """.stripMargin)
+
+ sql(
+ """
+ |CREATE TABLE table_one_pass (ID int, date Timestamp, country String,
+ |name String, phonetype String, serialname String, salary int)
+ |STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+
+ sql(
+ """
+ |LOAD DATA local inpath './src/test/resources/dataDiff.csv' INTO TABLE table_one_pass
+ |OPTIONS('DELIMITER'= ',', 'USE_KETTLE'='false', 'SINGLE_PASS'='true')
+ """.stripMargin)
+ }
+
+ test("test data loading use one pass") {
+ checkAnswer(
+ sql("select * from table_one_pass"),
+ sql("select * from table_two_pass")
+ )
+ }
+
+ test("test data loading use one pass when offer column dictionary file") {
+ sql(
+ """
+ |CREATE TABLE table_one_pass_2 (ID int, date Timestamp, country String,
+ |name String, phonetype String, serialname String, salary int)
+ |STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ """
+ |LOAD DATA local inpath './src/test/resources/dataDiff.csv' INTO TABLE table_one_pass_2
+ |OPTIONS('DELIMITER'= ',', 'USE_KETTLE'='false', 'SINGLE_PASS'='true', 'COLUMNDICT'=
+ |'country:./src/test/resources/columndictionary/country.csv, name:./src/test/resources/columndictionary/name.csv')
+ """.stripMargin)
+
+ checkAnswer(
+ sql("select * from table_one_pass_2"),
+ sql("select * from table_two_pass")
+ )
+ }
+
+ test("test data loading use one pass when do incremental load") {
+ sql(
+ """
+ |LOAD DATA local inpath './src/test/resources/dataIncrement.csv' INTO TABLE table_two_pass
+ |OPTIONS('DELIMITER'= ',', 'USE_KETTLE'='false', 'SINGLE_PASS'='false')
+ """.stripMargin)
+ sql(
+ """
+ |LOAD DATA local inpath './src/test/resources/dataIncrement.csv' INTO TABLE table_one_pass
+ |OPTIONS('DELIMITER'= ',', 'USE_KETTLE'='false', 'SINGLE_PASS'='true')
+ """.stripMargin)
+
+ checkAnswer(
+ sql("select * from table_one_pass"),
+ sql("select * from table_two_pass")
+ )
+ }
+
+ override def afterAll {
+ sql("DROP TABLE IF EXISTS table_two_pass")
+ sql("DROP TABLE IF EXISTS table_one_pass")
+ sql("DROP TABLE IF EXISTS table_one_pass_2")
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/05b26549/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/FilterProcessorTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/FilterProcessorTestCase.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/FilterProcessorTestCase.scala
index 9b3ac45..7d45e88 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/FilterProcessorTestCase.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/FilterProcessorTestCase.scala
@@ -178,7 +178,7 @@ class FilterProcessorTestCase extends QueryTest with BeforeAndAfterAll {
test("Greater Than equal to Filter with limit") {
checkAnswer(
- sql("select id from filtertestTables " + "where id >=999 limit 1"),
+ sql("select id from filtertestTables " + "where id >=999 order by id desc limit 1"),
Seq(Row(1000))
)
}