From 66fe760f99180872948c7ee9692695ea5b218cab Mon Sep 17 00:00:00 2001 From: anubhav100 Date: Sat, 27 May 2017 14:04:22 +0530 Subject: [PATCH] Resolved Bug For Select Count * Not Working In Hive --- .../hive/CarbonHiveRecordReader.java | 24 ++++++----- .../carbondata/hive/CarbonHiveSerDe.java | 13 +++--- .../hive/MapredCarbonInputFormat.java | 43 ++++++++++++++----- 3 files changed, 53 insertions(+), 27 deletions(-) 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 add4bafa55c..95743e70d94 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 void initialize(InputSplit inputSplit, Configuration conf) throws IOExcep } else { columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypeProperty); } + if (!colIds.equals("")) { + String[] arraySelectedColId = colIds.split(","); + List reqColTypes = new ArrayList(); - String[] arraySelectedColId = colIds.split(","); - List reqColTypes = new ArrayList(); - - 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 @@ 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: @@ -252,4 +256,4 @@ private Writable createObject(Object obj, ObjectInspector inspector) throws SerD } throw new SerDeException("Unknown data type" + inspector.getCategory()); } -} +} \ No newline at end of file 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 3a5c50a2d6d..2add00a3036 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 void initialize(@Nullable Configuration configuration, Properties tbl) } else { columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypeProperty); } - if (colIds != null) { + if (colIds != null && !colIds.equals("")) { reqColNames = new ArrayList(); String[] arraySelectedColId = colIds.split(","); @@ -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); @@ -135,7 +136,7 @@ public Class 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; @@ -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(); @@ -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; } @@ -246,4 +247,4 @@ public Object deserialize(Writable writable) throws SerDeException { public ObjectInspector getObjectInspector() throws SerDeException { return objInspector; } -} +} \ No newline at end of file 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 5caf5a8ea9b..5059b9177b9 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 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.Reporter; import org.apache.hadoop.mapreduce.Job; - public class MapredCarbonInputFormat extends CarbonInputFormat implements InputFormat, 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 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 getRecordReader(InputSplit inputSplit, JobConf jobConf, - Reporter reporter) throws IOException { + Reporter reporter) throws IOException { QueryModel queryModel = getQueryModel(jobConf); CarbonReadSupport readSupport = getReadSupportClass(jobConf); return new CarbonHiveRecordReader(queryModel, readSupport, inputSplit, jobConf); @@ -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 carbonDimensionList = carbonTable.getAllDimensions(); + List 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 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; } }