You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ch...@apache.org on 2017/08/17 11:01:04 UTC
carbondata git commit: [CARBONDATA-1359] Unable to use carbondata on
hive
Repository: carbondata
Updated Branches:
refs/heads/master 0e1d37e64 -> 27b178d98
[CARBONDATA-1359] Unable to use carbondata on hive
This closes #1231
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/27b178d9
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/27b178d9
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/27b178d9
Branch: refs/heads/master
Commit: 27b178d98777a019c0439192bf967e24acde358c
Parents: 0e1d37e
Author: shivangi <sh...@knoldus.in>
Authored: Fri Aug 4 13:25:02 2017 +0530
Committer: chenliang613 <ch...@apache.org>
Committed: Thu Aug 17 19:00:47 2017 +0800
----------------------------------------------------------------------
.../hive/MapredCarbonInputFormat.java | 59 ++++++++++----------
1 file changed, 30 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/27b178d9/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 5190846..273536a 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
@@ -52,32 +52,6 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
implements InputFormat<Void, ArrayWritable>, CombineHiveInputFormat.AvoidSplitCombination {
private static final String CARBON_TABLE = "mapreduce.input.carboninputformat.table";
- @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;
- 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());
- }
- return splits;
- }
-
- @Override
- public RecordReader<Void, ArrayWritable> getRecordReader(InputSplit inputSplit, JobConf jobConf,
- Reporter reporter) throws IOException {
- String path = null;
- if (inputSplit instanceof CarbonHiveInputSplit) {
- path = ((CarbonHiveInputSplit) inputSplit).getPath().toString();
- }
- QueryModel queryModel = getQueryModel(jobConf, path);
- CarbonReadSupport<ArrayWritable> readSupport = new CarbonDictionaryDecodeReadSupport<>();
- return new CarbonHiveRecordReader(queryModel, readSupport, inputSplit, jobConf);
- }
-
/**
* this method will read the schema from the physical file and populate into CARBON_TABLE
*
@@ -94,7 +68,7 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
} else {
if (paths != null) {
for (String inputPath : inputPaths) {
- if (paths.startsWith(inputPath)) {
+ if (paths.startsWith(inputPath.replace("file:", ""))) {
validInputPath = inputPath;
break;
}
@@ -106,6 +80,7 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
// read the schema file to get the absoluteTableIdentifier having the correct table id
// persisted in the schema
CarbonTable carbonTable = SchemaReader.readCarbonTableFromStore(absoluteTableIdentifier);
+ configuration.set(CARBON_TABLE, ObjectSerializationUtil.convertObjectToString(carbonTable));
setTableInfo(configuration, carbonTable.getTableInfo());
}
@@ -117,6 +92,32 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
return (CarbonTable) ObjectSerializationUtil.convertStringToObject(carbonTableStr);
}
+ @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;
+ 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());
+ }
+ return splits;
+ }
+
+ @Override
+ public RecordReader<Void, ArrayWritable> getRecordReader(InputSplit inputSplit, JobConf jobConf,
+ Reporter reporter) throws IOException {
+ String path = null;
+ if (inputSplit instanceof CarbonHiveInputSplit) {
+ path = ((CarbonHiveInputSplit) inputSplit).getPath().toString();
+ }
+ QueryModel queryModel = getQueryModel(jobConf, path);
+ CarbonReadSupport<ArrayWritable> readSupport = new CarbonDictionaryDecodeReadSupport<>();
+ return new CarbonHiveRecordReader(queryModel, readSupport, inputSplit, jobConf);
+ }
+
private QueryModel getQueryModel(Configuration configuration, String path) throws IOException {
CarbonTable carbonTable = getCarbonTable(configuration, path);
// getting the table absoluteTableIdentifier from the carbonTable
@@ -128,8 +129,8 @@ public class MapredCarbonInputFormat extends CarbonInputFormat<ArrayWritable>
String projection = getProjection(configuration, carbonTable,
identifier.getCarbonTableIdentifier().getTableName());
CarbonQueryPlan queryPlan = CarbonInputFormatUtil.createQueryPlan(carbonTable, projection);
- QueryModel queryModel = QueryModel.createModel(identifier, queryPlan, carbonTable,
- new DataTypeConverterImpl());
+ QueryModel queryModel =
+ QueryModel.createModel(identifier, queryPlan, carbonTable, new DataTypeConverterImpl());
// set the filter to the query model in order to filter blocklet before scan
Expression filter = getFilterPredicates(configuration);
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);