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 2017/06/20 07:29:18 UTC
[12/56] [abbrv] carbondata git commit: Resolved Bug For Select Count
* Not Working In Hive
Resolved Bug For Select Count * Not Working In Hive
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/0da86b6f
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/0da86b6f
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/0da86b6f
Branch: refs/heads/streaming_ingest
Commit: 0da86b6f72c4101ef02157f9623f9f5ea590b09d
Parents: b9b11fe
Author: anubhav100 <an...@knoldus.in>
Authored: Sat May 27 14:04:22 2017 +0530
Committer: chenliang613 <ch...@apache.org>
Committed: Wed Jun 14 11:34:59 2017 +0800
----------------------------------------------------------------------
.../carbondata/hive/CarbonHiveRecordReader.java | 24 ++++++-----
.../apache/carbondata/hive/CarbonHiveSerDe.java | 13 +++---
.../hive/MapredCarbonInputFormat.java | 43 +++++++++++++++-----
3 files changed, 53 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0da86b6f/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveRecordReader.java
----------------------------------------------------------------------
diff --git a/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveRecordReader.java b/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveRecordReader.java
index 81c3abf..42d9b68 100644
--- a/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveRecordReader.java
+++ b/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveRecordReader.java
@@ -111,16 +111,20 @@ public class CarbonHiveRecordReader extends CarbonRecordReader<ArrayWritable>
} else {
columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypeProperty);
}
+ if (!colIds.equals("")) {
+ String[] arraySelectedColId = colIds.split(",");
+ List<TypeInfo> reqColTypes = new ArrayList<TypeInfo>();
- String[] arraySelectedColId = colIds.split(",");
- List<TypeInfo> reqColTypes = new ArrayList<TypeInfo>();
-
- for (String anArrayColId : arraySelectedColId) {
- reqColTypes.add(columnTypes.get(Integer.parseInt(anArrayColId)));
+ for (String anArrayColId : arraySelectedColId) {
+ reqColTypes.add(columnTypes.get(Integer.parseInt(anArrayColId)));
+ }
+ // Create row related objects
+ rowTypeInfo = TypeInfoFactory.getStructTypeInfo(columnNames, reqColTypes);
+ this.objInspector = new CarbonObjectInspector((StructTypeInfo) rowTypeInfo);
+ } else {
+ rowTypeInfo = TypeInfoFactory.getStructTypeInfo(columnNames, columnTypes);
+ this.objInspector = new CarbonObjectInspector((StructTypeInfo) rowTypeInfo);
}
- // Create row related objects
- rowTypeInfo = TypeInfoFactory.getStructTypeInfo(columnNames, reqColTypes);
- this.objInspector = new CarbonObjectInspector((StructTypeInfo) rowTypeInfo);
}
@Override public boolean next(Void aVoid, ArrayWritable value) throws IOException {
@@ -227,7 +231,7 @@ public class CarbonHiveRecordReader extends CarbonRecordReader<ArrayWritable>
case LONG:
return new LongWritable((long) obj);
case SHORT:
- return new ShortWritable((Short) obj);
+ return new ShortWritable((short) obj);
case DATE:
return new DateWritable(new Date(Long.parseLong(String.valueOf(obj.toString()))));
case TIMESTAMP:
@@ -254,4 +258,4 @@ public class CarbonHiveRecordReader extends CarbonRecordReader<ArrayWritable>
}
throw new SerDeException("Unknown data type" + inspector.getCategory());
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0da86b6f/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveSerDe.java
----------------------------------------------------------------------
diff --git a/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveSerDe.java b/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveSerDe.java
index c6b0365..c8251eb 100644
--- a/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveSerDe.java
+++ b/integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveSerDe.java
@@ -102,7 +102,7 @@ public class CarbonHiveSerDe extends AbstractSerDe {
} else {
columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypeProperty);
}
- if (colIds != null) {
+ if (colIds != null && !colIds.equals("")) {
reqColNames = new ArrayList<String>();
String[] arraySelectedColId = colIds.split(",");
@@ -114,7 +114,8 @@ public class CarbonHiveSerDe extends AbstractSerDe {
// Create row related objects
rowTypeInfo = TypeInfoFactory.getStructTypeInfo(reqColNames, reqColTypes);
this.objInspector = new CarbonObjectInspector((StructTypeInfo) rowTypeInfo);
- } else {
+ }
+ else {
// Create row related objects
rowTypeInfo = TypeInfoFactory.getStructTypeInfo(columnNames, columnTypes);
this.objInspector = new CarbonObjectInspector((StructTypeInfo) rowTypeInfo);
@@ -135,7 +136,7 @@ public class CarbonHiveSerDe extends AbstractSerDe {
public Writable serialize(Object obj, ObjectInspector objectInspector) throws SerDeException {
if (!objInspector.getCategory().equals(ObjectInspector.Category.STRUCT)) {
throw new SerDeException("Cannot serialize " + objInspector.getCategory()
- + ". Can only serialize a struct");
+ + ". Can only serialize a struct");
}
serializedSize += ((StructObjectInspector) objInspector).getAllStructFieldRefs().size();
status = LAST_OPERATION.SERIALIZE;
@@ -156,7 +157,7 @@ public class CarbonHiveSerDe extends AbstractSerDe {
}
private ArrayWritable createArray(Object obj, ListObjectInspector inspector)
- throws SerDeException {
+ throws SerDeException {
List sourceArray = inspector.getList(obj);
ObjectInspector subInspector = inspector.getListElementObjectInspector();
List array = new ArrayList();
@@ -180,7 +181,7 @@ public class CarbonHiveSerDe extends AbstractSerDe {
}
private Writable createPrimitive(Object obj, PrimitiveObjectInspector inspector)
- throws SerDeException {
+ throws SerDeException {
if (obj == null) {
return null;
}
@@ -248,4 +249,4 @@ public class CarbonHiveSerDe extends AbstractSerDe {
public ObjectInspector getObjectInspector() throws SerDeException {
return objInspector;
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0da86b6f/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
----------------------------------------------------------------------
diff --git a/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java b/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
index 5caf5a8..5059b91 100644
--- a/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
+++ b/integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
@@ -21,6 +21,8 @@ import java.util.List;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.CarbonQueryPlan;
@@ -41,28 +43,26 @@ import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Job;
-
public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
implements InputFormat<Void, ArrayWritable>, CombineHiveInputFormat.AvoidSplitCombination {
- @Override
- public InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException {
+ @Override public InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException {
org.apache.hadoop.mapreduce.JobContext jobContext = Job.getInstance(jobConf);
List<org.apache.hadoop.mapreduce.InputSplit> splitList = super.getSplits(jobContext);
InputSplit[] splits = new InputSplit[splitList.size()];
CarbonInputSplit split = null;
for (int i = 0; i < splitList.size(); i++) {
split = (CarbonInputSplit) splitList.get(i);
- splits[i] = new CarbonHiveInputSplit(split.getSegmentId(), split.getPath(),
- split.getStart(), split.getLength(), split.getLocations(),
- split.getNumberOfBlocklets(), split.getVersion(), split.getBlockStorageIdMap());
+ splits[i] = new CarbonHiveInputSplit(split.getSegmentId(), split.getPath(), split.getStart(),
+ split.getLength(), split.getLocations(), split.getNumberOfBlocklets(), split.getVersion(),
+ split.getBlockStorageIdMap());
}
return splits;
}
@Override
public RecordReader<Void, ArrayWritable> getRecordReader(InputSplit inputSplit, JobConf jobConf,
- Reporter reporter) throws IOException {
+ Reporter reporter) throws IOException {
QueryModel queryModel = getQueryModel(jobConf);
CarbonReadSupport<ArrayWritable> readSupport = getReadSupportClass(jobConf);
return new CarbonHiveRecordReader(queryModel, readSupport, inputSplit, jobConf);
@@ -72,17 +72,39 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
CarbonTable carbonTable = getCarbonTable(configuration);
// getting the table absoluteTableIdentifier from the carbonTable
// to avoid unnecessary deserialization
+
+ String colNames = "";
AbsoluteTableIdentifier identifier = carbonTable.getAbsoluteTableIdentifier();
// query plan includes projection column
-
String projection = getColumnProjection(configuration);
if (projection == null) {
projection = configuration.get("hive.io.file.readcolumn.names");
}
+ if (projection.equals("")) {
+ List<CarbonDimension> carbonDimensionList = carbonTable.getAllDimensions();
+ List<CarbonMeasure> carbonMeasureList = carbonTable.getAllMeasures();
+
+ for (CarbonDimension aCarbonDimensionList : carbonDimensionList) {
+ colNames = (colNames + (aCarbonDimensionList.getColName())) + ",";
+ }
+ if (carbonMeasureList.size() < 1) {
+ colNames = colNames.substring(0, colNames.lastIndexOf(","));
+ }
+ for (int index = 0; index < carbonMeasureList.size(); index++) {
+ if (!carbonMeasureList.get(index).getColName().equals("default_dummy_measure")) {
+ if (index == carbonMeasureList.size() - 1) {
+ colNames = (colNames + (carbonMeasureList.get(index).getColName()));
+ } else {
+ colNames = (colNames + (carbonMeasureList.get(index).getColName())) + ",";
+ }
+ }
+ }
+ projection = colNames.trim();
+ configuration.set("hive.io.file.readcolumn.names", colNames);
+ }
CarbonQueryPlan queryPlan = CarbonInputFormatUtil.createQueryPlan(carbonTable, projection);
QueryModel queryModel = QueryModel.createModel(identifier, queryPlan, carbonTable);
-
// set the filter to the query model in order to filter blocklet before scan
Expression filter = getFilterPredicates(configuration);
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
@@ -92,8 +114,7 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
return queryModel;
}
- @Override
- public boolean shouldSkipCombine(Path path, Configuration conf) throws IOException {
+ @Override public boolean shouldSkipCombine(Path path, Configuration conf) throws IOException {
return true;
}
}