Skip to content

Commit

Permalink
[CARBONDATA-917] Resolved Bug for Select Count(*) not Working In Hive…
Browse files Browse the repository at this point in the history 
… This closes #963
  • Loading branch information
@chenliang613
chenliang613 committed Jun 14, 2017
2 parents b9b11fe + 0da86b6 commit 9eaaac5
Show file tree
Hide file tree
Showing 3 changed files with 53 additions and 27 deletions.
24 changes: 14 additions & 10 deletions integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveRecordReader.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,16 +111,20 @@ public void initialize(InputSplit inputSplit, Configuration conf) throws IOExcep
} 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 {
Expand Down Expand Up @@ -227,7 +231,7 @@ private Writable createPrimitive(Object obj, PrimitiveObjectInspector inspector)
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:
Expand All @@ -254,4 +258,4 @@ private Writable createObject(Object obj, ObjectInspector inspector) throws SerD
}
throw new SerDeException("Unknown data type" + inspector.getCategory());
}
}
}
13 changes: 7 additions & 6 deletions integration/hive/src/main/java/org/apache/carbondata/hive/CarbonHiveSerDe.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ public void initialize(@Nullable Configuration configuration, Properties tbl)
} else {
columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypeProperty);
}
if (colIds != null) {
if (colIds != null && !colIds.equals("")) {
reqColNames = new ArrayList<String>();

String[] arraySelectedColId = colIds.split(",");
Expand All @@ -114,7 +114,8 @@ public void initialize(@Nullable Configuration configuration, Properties tbl)
// 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);
Expand All @@ -135,7 +136,7 @@ public Class<? extends Writable> getSerializedClass() {
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;
Expand All @@ -156,7 +157,7 @@ public ArrayWritable createStruct(Object obj, StructObjectInspector inspector)
}

private ArrayWritable createArray(Object obj, ListObjectInspector inspector)
throws SerDeException {
throws SerDeException {
List sourceArray = inspector.getList(obj);
ObjectInspector subInspector = inspector.getListElementObjectInspector();
List array = new ArrayList();
Expand All @@ -180,7 +181,7 @@ private ArrayWritable createArray(Object obj, ListObjectInspector inspector)
}

private Writable createPrimitive(Object obj, PrimitiveObjectInspector inspector)
throws SerDeException {
throws SerDeException {
if (obj == null) {
return null;
}
Expand Down Expand Up @@ -248,4 +249,4 @@ public Object deserialize(Writable writable) throws SerDeException {
public ObjectInspector getObjectInspector() throws SerDeException {
return objInspector;
}
}
}
43 changes: 32 additions & 11 deletions integration/hive/src/main/java/org/apache/carbondata/hive/MapredCarbonInputFormat.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@

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;
Expand All @@ -41,28 +43,26 @@
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);
Expand All @@ -72,17 +72,39 @@ public QueryModel getQueryModel(Configuration configuration) throws IOException
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);
Expand All @@ -92,8 +114,7 @@ public QueryModel getQueryModel(Configuration configuration) throws IOException
return queryModel;
}

@Override
public boolean shouldSkipCombine(Path path, Configuration conf) throws IOException {
@Override public boolean shouldSkipCombine(Path path, Configuration conf) throws IOException {
return true;
}
}

0 comments on commit 9eaaac5

Please sign in to comment.