9518

Build # 9518

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/256364e8546db18000907e5ea90af711998d92fc">256364e85<a href="https://github.com/apache/carbondata/commit/256364e8546db18000907e5ea90af711998d92fc">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/8687a074427054a12c0d695f6725846a7a850067">8687a0744</a>">8687a0744</a><a href="https://github.com/apache/carbondata/commit/256364e8546db18000907e5ea90af711998d92fc"> into 934216de1">934216de1</a>

Run Details

178 of 178 new or added lines in 9 files covered. (100.0%)

62106 of 77375 relevant lines covered (80.27%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.42

/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonVectorizedRecordReader.java

/*
 * 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.hadoop.util;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datastore.FileReader;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.datatype.DecimalType;
import org.apache.carbondata.core.metadata.datatype.StructField;
import org.apache.carbondata.core.scan.executor.QueryExecutor;
import org.apache.carbondata.core.scan.executor.QueryExecutorFactory;
import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
import org.apache.carbondata.core.scan.model.ProjectionDimension;
import org.apache.carbondata.core.scan.model.ProjectionMeasure;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.core.scan.result.iterator.AbstractDetailQueryResultIterator;
import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.hadoop.AbstractRecordReader;
import org.apache.carbondata.hadoop.CarbonInputSplit;

import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.log4j.Logger;

/**
 * A specialized RecordReader that reads into CarbonColumnarBatches directly using the
 * carbondata column APIs and fills the data directly into columns.
 */
public class CarbonVectorizedRecordReader extends AbstractRecordReader<Object> {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CarbonVectorizedRecordReader.class.getName());

  private CarbonColumnarBatch carbonColumnarBatch;

  private QueryExecutor queryExecutor;

  private int batchIdx = 0;

  private int numBatched = 0;

  private AbstractDetailQueryResultIterator iterator;

  private QueryModel queryModel;

  public CarbonVectorizedRecordReader(QueryModel queryModel) {
    this.queryModel = queryModel;
  }

  @Override public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
      throws IOException, InterruptedException {
    List<CarbonInputSplit> splitList;
    if (inputSplit instanceof CarbonInputSplit) {
      // Read the footer offset and set.
      String splitPath = ((CarbonInputSplit) inputSplit).getPath().toString();
      if (((CarbonInputSplit) inputSplit).getDetailInfo().getBlockFooterOffset() == 0L) {
        FileReader reader = FileFactory.getFileHolder(FileFactory.getFileType(splitPath),
            taskAttemptContext.getConfiguration());
        ByteBuffer buffer = reader
            .readByteBuffer(FileFactory.getUpdatedFilePath(splitPath),
                ((CarbonInputSplit) inputSplit).getDetailInfo().getBlockSize() - 8,
                8);
        ((CarbonInputSplit) inputSplit).getDetailInfo().setBlockFooterOffset(buffer.getLong());
      }
      splitList = new ArrayList<>(1);
      splitList.add((CarbonInputSplit) inputSplit);
    } else {
      throw new RuntimeException("unsupported input split type: " + inputSplit);
    }
    List<TableBlockInfo> tableBlockInfoList = CarbonInputSplit.createBlocks(splitList);
    queryModel.setTableBlockInfos(tableBlockInfoList);
    queryModel.setVectorReader(true);
    try {
      queryExecutor =
          QueryExecutorFactory.getQueryExecutor(queryModel, taskAttemptContext.getConfiguration());
      iterator = (AbstractDetailQueryResultIterator) queryExecutor.execute(queryModel);
      initBatch();
    } catch (QueryExecutionException e) {
      LOGGER.error(e);
      throw new InterruptedException(e.getMessage());
    } catch (Exception e) {
      LOGGER.error(e);
      throw e;
    }
  }

  @Override public boolean nextKeyValue() throws IOException, InterruptedException {
    if (batchIdx >= numBatched) {
      if (!nextBatch()) return false;
    }
    ++batchIdx;
    return true;
  }


  private boolean nextBatch() {
    carbonColumnarBatch.reset();
    if (iterator.hasNext()) {
      iterator.processNextBatch(carbonColumnarBatch);
      numBatched = carbonColumnarBatch.getActualSize();
      batchIdx = 0;
      return true;
    }
    return false;
  }

  private void initBatch() {
    if (carbonColumnarBatch == null) {
      List<ProjectionDimension> queryDimension = queryModel.getProjectionDimensions();
      List<ProjectionMeasure> queryMeasures = queryModel.getProjectionMeasures();
      StructField[] fields = new StructField[queryDimension.size() + queryMeasures.size()];
      for (ProjectionDimension dim : queryDimension) {
        fields[dim.getOrdinal()] =
            new StructField(dim.getColumnName(), dim.getDimension().getDataType());
      }
      for (ProjectionMeasure msr : queryMeasures) {
        DataType dataType = msr.getMeasure().getDataType();
        if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.SHORT
            || dataType == DataTypes.INT || dataType == DataTypes.LONG
            || dataType == DataTypes.FLOAT || dataType == DataTypes.BYTE) {
          fields[msr.getOrdinal()] =
              new StructField(msr.getColumnName(), msr.getMeasure().getDataType());
        } else if (DataTypes.isDecimal(dataType)) {
          fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),
              new DecimalType(msr.getMeasure().getPrecision(), msr.getMeasure().getScale()));
        } else {
          fields[msr.getOrdinal()] = new StructField(msr.getColumnName(), DataTypes.DOUBLE);
        }
      }
      CarbonColumnVector[] vectors = new CarbonColumnVector[fields.length];
      for (int i = 0; i < fields.length; i++) {
        vectors[i] = new CarbonColumnVectorImpl(
            CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT,
            fields[i].getDataType());
      }
      carbonColumnarBatch = new CarbonColumnarBatch(vectors,
          CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT,
          new boolean[] {});
    }
  }

  @Override
  public Object getCurrentValue() throws IOException, InterruptedException {
    rowCount += 1;
    Object[] row = new Object[carbonColumnarBatch.columnVectors.length];
    for (int i = 0; i < carbonColumnarBatch.columnVectors.length; i ++) {
      if (carbonColumnarBatch.columnVectors[i].getType() == DataTypes.STRING
          || carbonColumnarBatch.columnVectors[i].getType() == DataTypes.VARCHAR) {
        byte[] data = (byte[]) carbonColumnarBatch.columnVectors[i].getData(batchIdx - 1);
        row[i] = ByteUtil.toString(data, 0, data.length);
      } else if (carbonColumnarBatch.columnVectors[i].getType() == DataTypes.BOOLEAN) {
        byte data = (byte) carbonColumnarBatch.columnVectors[i].getData(batchIdx - 1);
        row[i] = ByteUtil.toBoolean(data);
      } else {
        row[i] = carbonColumnarBatch.columnVectors[i].getData(batchIdx - 1);
      }
    }
    return row;
  }

  @Override public Void getCurrentKey() throws IOException, InterruptedException {
    throw new UnsupportedOperationException("Operation not allowed on CarbonVectorizedReader");
  }

  @Override public float getProgress() throws IOException, InterruptedException {
    // TODO : Implement it based on total number of rows it is going to retrieve.
    return 0;
  }

  @Override public void close() throws IOException {
    logStatistics(rowCount, queryModel.getStatisticsRecorder());
    if (carbonColumnarBatch != null) {
      carbonColumnarBatch = null;
    }
    if (iterator != null) {
      iterator.close();
    }
    try {
      queryExecutor.finish();
    } catch (QueryExecutionException e) {
      throw new IOException(e);
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.hadoop.util;
19
20	import java.io.IOException;
21	import java.nio.ByteBuffer;
22	import java.util.ArrayList;
23	import java.util.List;
24
25	import org.apache.carbondata.common.logging.LogServiceFactory;
26	import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
27	import org.apache.carbondata.core.datastore.FileReader;
28	import org.apache.carbondata.core.datastore.block.TableBlockInfo;
29	import org.apache.carbondata.core.datastore.impl.FileFactory;
30	import org.apache.carbondata.core.metadata.datatype.DataType;
31	import org.apache.carbondata.core.metadata.datatype.DataTypes;
32	import org.apache.carbondata.core.metadata.datatype.DecimalType;
33	import org.apache.carbondata.core.metadata.datatype.StructField;
34	import org.apache.carbondata.core.scan.executor.QueryExecutor;
35	import org.apache.carbondata.core.scan.executor.QueryExecutorFactory;
36	import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
37	import org.apache.carbondata.core.scan.model.ProjectionDimension;
38	import org.apache.carbondata.core.scan.model.ProjectionMeasure;
39	import org.apache.carbondata.core.scan.model.QueryModel;
40	import org.apache.carbondata.core.scan.result.iterator.AbstractDetailQueryResultIterator;
41	import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
42	import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
43	import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
44	import org.apache.carbondata.core.util.ByteUtil;
45	import org.apache.carbondata.hadoop.AbstractRecordReader;
46	import org.apache.carbondata.hadoop.CarbonInputSplit;
47
48	import org.apache.hadoop.mapreduce.InputSplit;
49	import org.apache.hadoop.mapreduce.TaskAttemptContext;
50	import org.apache.log4j.Logger;
51
52	/**
53	* A specialized RecordReader that reads into CarbonColumnarBatches directly using the
54	* carbondata column APIs and fills the data directly into columns.
55	*/
56	public class CarbonVectorizedRecordReader extends AbstractRecordReader<Object> {
57
58	private static final Logger LOGGER =	1✔
59	LogServiceFactory.getLogService(CarbonVectorizedRecordReader.class.getName());	1✔
60
61	private CarbonColumnarBatch carbonColumnarBatch;
62
63	private QueryExecutor queryExecutor;
64
65	private int batchIdx = 0;	1✔
66
67	private int numBatched = 0;	1✔
68
69	private AbstractDetailQueryResultIterator iterator;
70
71	private QueryModel queryModel;
72
73	public CarbonVectorizedRecordReader(QueryModel queryModel) {	1✔
74	this.queryModel = queryModel;	1✔
75	}	1✔
76
77	@Override public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
78	throws IOException, InterruptedException {
79	List<CarbonInputSplit> splitList;
80	if (inputSplit instanceof CarbonInputSplit) {	1✔
81	// Read the footer offset and set.
82	String splitPath = ((CarbonInputSplit) inputSplit).getPath().toString();	1✔
83	if (((CarbonInputSplit) inputSplit).getDetailInfo().getBlockFooterOffset() == 0L) {	1✔
84	FileReader reader = FileFactory.getFileHolder(FileFactory.getFileType(splitPath),	1✔
85	taskAttemptContext.getConfiguration());	1✔
86	ByteBuffer buffer = reader	1✔
87	.readByteBuffer(FileFactory.getUpdatedFilePath(splitPath),	1✔
88	((CarbonInputSplit) inputSplit).getDetailInfo().getBlockSize() - 8,	1✔
89	8);
90	((CarbonInputSplit) inputSplit).getDetailInfo().setBlockFooterOffset(buffer.getLong());	1✔
91	}
92	splitList = new ArrayList<>(1);	1✔
93	splitList.add((CarbonInputSplit) inputSplit);	1✔
94	} else {	1✔
95	throw new RuntimeException("unsupported input split type: " + inputSplit);	×
96	}
97	List<TableBlockInfo> tableBlockInfoList = CarbonInputSplit.createBlocks(splitList);	1✔
98	queryModel.setTableBlockInfos(tableBlockInfoList);	1✔
99	queryModel.setVectorReader(true);	1✔
100	try {
101	queryExecutor =	1✔
102	QueryExecutorFactory.getQueryExecutor(queryModel, taskAttemptContext.getConfiguration());	1✔
103	iterator = (AbstractDetailQueryResultIterator) queryExecutor.execute(queryModel);	1✔
104	initBatch();	1✔
105	} catch (QueryExecutionException e) {	×
106	LOGGER.error(e);	×
107	throw new InterruptedException(e.getMessage());	×
108	} catch (Exception e) {	×
109	LOGGER.error(e);	×
110	throw e;	×
111	}	1✔
112	}	1✔
113
114	@Override public boolean nextKeyValue() throws IOException, InterruptedException {
115	if (batchIdx >= numBatched) {	1✔
116	if (!nextBatch()) return false;	1✔
117	}
118	++batchIdx;	1✔
119	return true;	1✔
120	}
121
122
123	private boolean nextBatch() {
124	carbonColumnarBatch.reset();	1✔
125	if (iterator.hasNext()) {	1✔
126	iterator.processNextBatch(carbonColumnarBatch);	1✔
127	numBatched = carbonColumnarBatch.getActualSize();	1✔
128	batchIdx = 0;	1✔
129	return true;	1✔
130	}
131	return false;	1✔
132	}
133
134	private void initBatch() {
135	if (carbonColumnarBatch == null) {	1✔
136	List<ProjectionDimension> queryDimension = queryModel.getProjectionDimensions();	1✔
137	List<ProjectionMeasure> queryMeasures = queryModel.getProjectionMeasures();	1✔
138	StructField[] fields = new StructField[queryDimension.size() + queryMeasures.size()];	1✔
139	for (ProjectionDimension dim : queryDimension) {	1✔
140	fields[dim.getOrdinal()] =	1✔
141	new StructField(dim.getColumnName(), dim.getDimension().getDataType());	1✔
142	}	1✔
143	for (ProjectionMeasure msr : queryMeasures) {	1✔
144	DataType dataType = msr.getMeasure().getDataType();	1✔
145	if (dataType == DataTypes.BOOLEAN \|\| dataType == DataTypes.SHORT	1✔
146	\|\| dataType == DataTypes.INT \|\| dataType == DataTypes.LONG
147	\|\| dataType == DataTypes.FLOAT \|\| dataType == DataTypes.BYTE) {
148	fields[msr.getOrdinal()] =	1✔
149	new StructField(msr.getColumnName(), msr.getMeasure().getDataType());	1✔
150	} else if (DataTypes.isDecimal(dataType)) {	1✔
151	fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),	1✔
152	new DecimalType(msr.getMeasure().getPrecision(), msr.getMeasure().getScale()));	1✔
153	} else {
154	fields[msr.getOrdinal()] = new StructField(msr.getColumnName(), DataTypes.DOUBLE);	1✔
155	}
156	}	1✔
157	CarbonColumnVector[] vectors = new CarbonColumnVector[fields.length];	1✔
158	for (int i = 0; i < fields.length; i++) {	1✔
159	vectors[i] = new CarbonColumnVectorImpl(	1✔
160	CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT,
161	fields[i].getDataType());	1✔
162	}
163	carbonColumnarBatch = new CarbonColumnarBatch(vectors,	1✔
164	CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT,
165	new boolean[] {});
166	}
167	}	1✔
168
169	@Override
170	public Object getCurrentValue() throws IOException, InterruptedException {
171	rowCount += 1;	1✔
172	Object[] row = new Object[carbonColumnarBatch.columnVectors.length];	1✔
173	for (int i = 0; i < carbonColumnarBatch.columnVectors.length; i ++) {	1✔
174	if (carbonColumnarBatch.columnVectors[i].getType() == DataTypes.STRING	1✔
175	\|\| carbonColumnarBatch.columnVectors[i].getType() == DataTypes.VARCHAR) {	1✔
176	byte[] data = (byte[]) carbonColumnarBatch.columnVectors[i].getData(batchIdx - 1);	1✔
177	row[i] = ByteUtil.toString(data, 0, data.length);	1✔
178	} else if (carbonColumnarBatch.columnVectors[i].getType() == DataTypes.BOOLEAN) {	1✔
179	byte data = (byte) carbonColumnarBatch.columnVectors[i].getData(batchIdx - 1);	1✔
180	row[i] = ByteUtil.toBoolean(data);	1✔
181	} else {	1✔
182	row[i] = carbonColumnarBatch.columnVectors[i].getData(batchIdx - 1);	1✔
183	}
184	}
185	return row;	1✔
186	}
187
188	@Override public Void getCurrentKey() throws IOException, InterruptedException {
189	throw new UnsupportedOperationException("Operation not allowed on CarbonVectorizedReader");	×
190	}
191
192	@Override public float getProgress() throws IOException, InterruptedException {
193	// TODO : Implement it based on total number of rows it is going to retrieve.
194	return 0;	×
195	}
196
197	@Override public void close() throws IOException {
198	logStatistics(rowCount, queryModel.getStatisticsRecorder());	1✔
199	if (carbonColumnarBatch != null) {	1✔
200	carbonColumnarBatch = null;	1✔
201	}
202	if (iterator != null) {	1✔
203	iterator.close();	1✔
204	}
205	try {
206	queryExecutor.finish();	1✔
207	} catch (QueryExecutionException e) {	×
208	throw new IOException(e);	×
209	}	1✔
210	}	1✔
211	}

apache / carbondata / 9518

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous