9614

Build # 9614

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc2d0ed9a1">9cd4deb9d">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc2d0ed9a1">9cd4deb9d</a><a href="https://github.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc2d0ed9a1">&lt;a href=&quot;https://github.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc2d0ed9a1&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc2d0ed9a1&amp;quot;&amp;gt;9cd4deb9d&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc2d0ed9a1&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/9cd4deb9defe8483b9b4bb653f7577bc&amp;lt;/a&amp;gt;2d0ed</a>9a1&quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/4990e1d5f&quot;&gt;4990e1d... (continued)

Run Details

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

62264 of 77513 relevant lines covered (80.33%)

1.06 hits per line

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

73.68

/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.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.spark.vectorreader;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.cache.dictionary.Dictionary;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryGenerator;
import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
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.util.CarbonUtil;
import org.apache.carbondata.hadoop.AbstractRecordReader;
import org.apache.carbondata.hadoop.CarbonInputSplit;
import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
import org.apache.carbondata.hadoop.InputMetricsStats;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.spark.memory.MemoryMode;
import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.execution.vectorized.ColumnVectorUtils;
import org.apache.spark.sql.CarbonVectorProxy;
import org.apache.spark.sql.types.DecimalType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

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

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

  private int batchIdx = 0;

  private static final int DEFAULT_BATCH_SIZE = 4 * 1024;

  private int numBatched = 0;

  private CarbonVectorProxy vectorProxy;

  private CarbonColumnarBatch carbonColumnarBatch;

  /**
   * If true, this class returns batches instead of rows.
   */
  private boolean returnColumnarBatch;

  private boolean[] isNoDictStringField;

  /**
   * The default config on whether columnarBatch should be onheap.
   */
  private static final MemoryMode DEFAULT_MEMORY_MODE = MemoryMode.ON_HEAP;

  private QueryModel queryModel;

  private AbstractDetailQueryResultIterator iterator;

  private QueryExecutor queryExecutor;

  private InputMetricsStats inputMetricsStats;

  public VectorizedCarbonRecordReader(QueryModel queryModel, InputMetricsStats inputMetricsStats,
      String enableBatch) {
    this.queryModel = queryModel;
    this.inputMetricsStats = inputMetricsStats;
    if (enableBatch.equals("true")) {
      enableReturningBatches();
    }
  }


  /*
 * Can be called before any rows are returned to enable returning columnar batches directly.
 */
  public void enableReturningBatches() {
    returnColumnarBatch = true;
  }

  /**
   * Implementation of RecordReader API.
   */
  @Override
  public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
      throws IOException, InterruptedException, UnsupportedOperationException {
    // The input split can contain single HDFS block or multiple blocks, so firstly get all the
    // blocks and then set them in the query model.
    List<CarbonInputSplit> splitList;
    if (inputSplit instanceof CarbonInputSplit) {
      splitList = new ArrayList<>(1);
      splitList.add((CarbonInputSplit) inputSplit);
    } else if (inputSplit instanceof CarbonMultiBlockSplit) {
      // contains multiple blocks, this is an optimization for concurrent query.
      CarbonMultiBlockSplit multiBlockSplit = (CarbonMultiBlockSplit) inputSplit;
      splitList = multiBlockSplit.getAllSplits();
    } 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);
    } catch (QueryExecutionException e) {
      if (ExceptionUtils.indexOfThrowable(e, FileNotFoundException.class) > 0) {
        LOGGER.error(e);
        throw new InterruptedException(
            "Insert overwrite may be in progress.Please check " + e.getMessage());
      }
      throw new InterruptedException(e.getMessage());
    } catch (Exception e) {
      if (ExceptionUtils.indexOfThrowable(e, FileNotFoundException.class) > 0) {
        LOGGER.error(e);
        throw new InterruptedException(
            "Insert overwrite may be in progress.Please check " + e.getMessage());
      }
      throw e;
    }
  }

  @Override
  public void close() throws IOException {
    logStatistics(rowCount, queryModel.getStatisticsRecorder());
    if (vectorProxy != null) {
      vectorProxy.close();
      vectorProxy = null;
    }
    // clear dictionary cache
    Map<String, Dictionary> columnToDictionaryMapping = queryModel.getColumnToDictionaryMapping();
    if (null != columnToDictionaryMapping) {
      for (Map.Entry<String, Dictionary> entry : columnToDictionaryMapping.entrySet()) {
        CarbonUtil.clearDictionaryCache(entry.getValue());
      }
    }
    try {
      queryExecutor.finish();
    } catch (QueryExecutionException e) {
      throw new IOException(e);
    }
  }

  @Override
  public boolean nextKeyValue() throws IOException, InterruptedException {
    resultBatch();

    if (returnColumnarBatch) {
      return nextBatch();
    }

    if (batchIdx >= numBatched) {
      if (!nextBatch()) return false;
    }
    ++batchIdx;
    return true;
  }

  @Override
  public Object getCurrentValue() throws IOException, InterruptedException {
    if (returnColumnarBatch) {
      int value = vectorProxy.numRows();
      rowCount += value;
      if (inputMetricsStats != null) {
        inputMetricsStats.incrementRecordRead((long) value);
      }
      return vectorProxy.getColumnarBatch();
    }
    rowCount += 1;
    return vectorProxy.getRow(batchIdx - 1);
  }

  @Override
  public Void getCurrentKey() throws IOException, InterruptedException {
    return null;
  }

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

  /**
   * Returns the ColumnarBatch object that will be used for all rows returned by this reader.
   * This object is reused. Calling this enables the vectorized reader. This should be called
   * before any calls to nextKeyValue/nextBatch.
   */

  public void initBatch(MemoryMode memMode, StructType partitionColumns,
      InternalRow partitionValues) {
    List<ProjectionDimension> queryDimension = queryModel.getProjectionDimensions();
    List<ProjectionMeasure> queryMeasures = queryModel.getProjectionMeasures();
    StructField[] fields = new StructField[queryDimension.size() + queryMeasures.size()];
    this.isNoDictStringField = new boolean[queryDimension.size() + queryMeasures.size()];
    for (int i = 0; i < queryDimension.size(); i++) {
      ProjectionDimension dim = queryDimension.get(i);
      if (dim.getDimension().hasEncoding(Encoding.DIRECT_DICTIONARY)) {
        DirectDictionaryGenerator generator = DirectDictionaryKeyGeneratorFactory
            .getDirectDictionaryGenerator(dim.getDimension().getDataType());
        fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),
            CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(generator.getReturnType()), true, null);
      } else if (!dim.getDimension().hasEncoding(Encoding.DICTIONARY)) {
        if (dim.getDimension().getDataType() == DataTypes.STRING
            || dim.getDimension().getDataType() == DataTypes.VARCHAR || dim.getDimension()
            .getColumnSchema().isLocalDictColumn()) {
          this.isNoDictStringField[dim.getOrdinal()] = true;
        }
        fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),
            CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(dim.getDimension().getDataType()), true,
            null);
      } else if (dim.getDimension().isComplex()) {
        fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),
            CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(dim.getDimension().getDataType()), true,
            null);
      } else {
        fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),
            CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(DataTypes.INT), true, null);
      }
    }

    for (int i = 0; i < queryMeasures.size(); i++) {
      ProjectionMeasure msr = queryMeasures.get(i);
      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(),
            CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(msr.getMeasure().getDataType()), true,
            null);
      } else if (DataTypes.isDecimal(dataType)) {
        fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),
            new DecimalType(msr.getMeasure().getPrecision(), msr.getMeasure().getScale()), true,
            null);
      } else {
        fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),
            CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(DataTypes.DOUBLE), true, null);
      }
    }
    StructType schema = new StructType(fields);
    if (partitionColumns != null) {
      for (StructField field : partitionColumns.fields()) {
        schema = schema.add(field);
      }
    }
    short batchSize = DEFAULT_BATCH_SIZE;
    if (queryModel.isDirectVectorFill()) {
      batchSize = CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT;
    }
    vectorProxy = new CarbonVectorProxy(DEFAULT_MEMORY_MODE, schema, batchSize);

    if (partitionColumns != null) {
      int partitionIdx = fields.length;
      for (int i = 0; i < partitionColumns.fields().length; i++) {
        ColumnVectorUtils.populate(vectorProxy.column(i + partitionIdx), partitionValues, i);
        vectorProxy.column(i + partitionIdx).setIsConstant();
      }
    }
    CarbonColumnVector[] vectors = new CarbonColumnVector[fields.length];
    boolean[] filteredRows = null;
    if (queryModel.isDirectVectorFill()) {
      for (int i = 0; i < fields.length; i++) {
        vectors[i] = new ColumnarVectorWrapperDirect(vectorProxy, i);
        if (isNoDictStringField[i]) {
          ((ColumnarVectorWrapperDirect) vectors[i]).reserveDictionaryIds();
        }
      }
    } else {
      filteredRows = new boolean[vectorProxy.numRows()];
      for (int i = 0; i < fields.length; i++) {
        vectors[i] = new ColumnarVectorWrapper(vectorProxy, i);
        if (isNoDictStringField[i]) {
          if (vectors[i] instanceof ColumnarVectorWrapper) {
            ((ColumnarVectorWrapper) vectors[i]).reserveDictionaryIds();
          }
        }
      }
    }
    carbonColumnarBatch = new CarbonColumnarBatch(vectors, vectorProxy.numRows(), filteredRows);
  }

  private void initBatch() {
    initBatch(DEFAULT_MEMORY_MODE, new StructType(), InternalRow.empty());
  }

  private void resultBatch() {
    if (vectorProxy == null) initBatch();
  }



  /**
   * Advances to the next batch of rows. Returns false if there are no more.
   */
  private boolean nextBatch() {
    if (null != isNoDictStringField) {
      for (int i = 0; i < isNoDictStringField.length; i++) {
        if (isNoDictStringField[i]) {
          vectorProxy.resetDictionaryIds(i);
        }
      }
    }
    vectorProxy.reset();
    carbonColumnarBatch.reset();
    if (iterator.hasNext()) {
      iterator.processNextBatch(carbonColumnarBatch);
      int actualSize = carbonColumnarBatch.getActualSize();
      vectorProxy.setNumRows(actualSize);
      numBatched = actualSize;
      batchIdx = 0;
      return true;
    }
    return false;
  }

}

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.spark.vectorreader;
19
20	import java.io.FileNotFoundException;
21	import java.io.IOException;
22	import java.util.ArrayList;
23	import java.util.List;
24	import java.util.Map;
25
26	import org.apache.log4j.Logger;
27	import org.apache.carbondata.common.logging.LogServiceFactory;
28	import org.apache.carbondata.core.cache.dictionary.Dictionary;
29	import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
30	import org.apache.carbondata.core.datastore.block.TableBlockInfo;
31	import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryGenerator;
32	import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
33	import org.apache.carbondata.core.metadata.datatype.DataType;
34	import org.apache.carbondata.core.metadata.datatype.DataTypes;
35	import org.apache.carbondata.core.metadata.encoder.Encoding;
36	import org.apache.carbondata.core.scan.executor.QueryExecutor;
37	import org.apache.carbondata.core.scan.executor.QueryExecutorFactory;
38	import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
39	import org.apache.carbondata.core.scan.model.ProjectionDimension;
40	import org.apache.carbondata.core.scan.model.ProjectionMeasure;
41	import org.apache.carbondata.core.scan.model.QueryModel;
42	import org.apache.carbondata.core.scan.result.iterator.AbstractDetailQueryResultIterator;
43	import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
44	import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
45	import org.apache.carbondata.core.util.CarbonUtil;
46	import org.apache.carbondata.hadoop.AbstractRecordReader;
47	import org.apache.carbondata.hadoop.CarbonInputSplit;
48	import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
49	import org.apache.carbondata.hadoop.InputMetricsStats;
50
51	import org.apache.commons.lang3.exception.ExceptionUtils;
52	import org.apache.hadoop.mapreduce.InputSplit;
53	import org.apache.hadoop.mapreduce.TaskAttemptContext;
54	import org.apache.spark.memory.MemoryMode;
55	import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil;
56	import org.apache.spark.sql.catalyst.InternalRow;
57	import org.apache.spark.sql.execution.vectorized.ColumnVectorUtils;
58	import org.apache.spark.sql.CarbonVectorProxy;
59	import org.apache.spark.sql.types.DecimalType;
60	import org.apache.spark.sql.types.StructField;
61	import org.apache.spark.sql.types.StructType;
62
63	/**
64	* A specialized RecordReader that reads into InternalRows or ColumnarBatches directly using the
65	* carbondata column APIs and fills the data directly into columns.
66	*/
67	public class VectorizedCarbonRecordReader extends AbstractRecordReader<Object> {
68
69	private static final Logger LOGGER =	1✔
70	LogServiceFactory.getLogService(VectorizedCarbonRecordReader.class.getName());	1✔
71
72	private int batchIdx = 0;	1✔
73
74	private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
75
76	private int numBatched = 0;	1✔
77
78	private CarbonVectorProxy vectorProxy;
79
80	private CarbonColumnarBatch carbonColumnarBatch;
81
82	/**
83	* If true, this class returns batches instead of rows.
84	*/
85	private boolean returnColumnarBatch;
86
87	private boolean[] isNoDictStringField;
88
89	/**
90	* The default config on whether columnarBatch should be onheap.
91	*/
92	private static final MemoryMode DEFAULT_MEMORY_MODE = MemoryMode.ON_HEAP;	1✔
93
94	private QueryModel queryModel;
95
96	private AbstractDetailQueryResultIterator iterator;
97
98	private QueryExecutor queryExecutor;
99
100	private InputMetricsStats inputMetricsStats;
101
102	public VectorizedCarbonRecordReader(QueryModel queryModel, InputMetricsStats inputMetricsStats,
103	String enableBatch) {	1✔
104	this.queryModel = queryModel;	1✔
105	this.inputMetricsStats = inputMetricsStats;	1✔
106	if (enableBatch.equals("true")) {	1✔
107	enableReturningBatches();	1✔
108	}
109	}	1✔
110
111
112	/*
113	* Can be called before any rows are returned to enable returning columnar batches directly.
114	*/
115	public void enableReturningBatches() {
116	returnColumnarBatch = true;	1✔
117	}	1✔
118
119	/**
120	* Implementation of RecordReader API.
121	*/
122	@Override
123	public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
124	throws IOException, InterruptedException, UnsupportedOperationException {
125	// The input split can contain single HDFS block or multiple blocks, so firstly get all the
126	// blocks and then set them in the query model.
127	List<CarbonInputSplit> splitList;
128	if (inputSplit instanceof CarbonInputSplit) {	1✔
129	splitList = new ArrayList<>(1);	1✔
130	splitList.add((CarbonInputSplit) inputSplit);	1✔
131	} else if (inputSplit instanceof CarbonMultiBlockSplit) {	×
132	// contains multiple blocks, this is an optimization for concurrent query.
133	CarbonMultiBlockSplit multiBlockSplit = (CarbonMultiBlockSplit) inputSplit;	×
134	splitList = multiBlockSplit.getAllSplits();	×
135	} else {	×
136	throw new RuntimeException("unsupported input split type: " + inputSplit);	×
137	}
138	List<TableBlockInfo> tableBlockInfoList = CarbonInputSplit.createBlocks(splitList);	1✔
139	queryModel.setTableBlockInfos(tableBlockInfoList);	1✔
140	queryModel.setVectorReader(true);	1✔
141	try {
142	queryExecutor =	1✔
143	QueryExecutorFactory.getQueryExecutor(queryModel, taskAttemptContext.getConfiguration());	1✔
144	iterator = (AbstractDetailQueryResultIterator) queryExecutor.execute(queryModel);	1✔
145	} catch (QueryExecutionException e) {	×
146	if (ExceptionUtils.indexOfThrowable(e, FileNotFoundException.class) > 0) {	×
147	LOGGER.error(e);	×
148	throw new InterruptedException(	×
149	"Insert overwrite may be in progress.Please check " + e.getMessage());	×
150	}
151	throw new InterruptedException(e.getMessage());	×
152	} catch (Exception e) {	1✔
153	if (ExceptionUtils.indexOfThrowable(e, FileNotFoundException.class) > 0) {	1✔
154	LOGGER.error(e);	×
155	throw new InterruptedException(	×
156	"Insert overwrite may be in progress.Please check " + e.getMessage());	×
157	}
158	throw e;	1✔
159	}	1✔
160	}	1✔
161
162	@Override
163	public void close() throws IOException {
164	logStatistics(rowCount, queryModel.getStatisticsRecorder());	1✔
165	if (vectorProxy != null) {	1✔
166	vectorProxy.close();	1✔
167	vectorProxy = null;	1✔
168	}
169	// clear dictionary cache
170	Map<String, Dictionary> columnToDictionaryMapping = queryModel.getColumnToDictionaryMapping();	1✔
171	if (null != columnToDictionaryMapping) {	1✔
172	for (Map.Entry<String, Dictionary> entry : columnToDictionaryMapping.entrySet()) {	1✔
173	CarbonUtil.clearDictionaryCache(entry.getValue());	×
174	}	×
175	}
176	try {
177	queryExecutor.finish();	1✔
178	} catch (QueryExecutionException e) {	×
179	throw new IOException(e);	×
180	}	1✔
181	}	1✔
182
183	@Override
184	public boolean nextKeyValue() throws IOException, InterruptedException {
185	resultBatch();	1✔
186
187	if (returnColumnarBatch) {	1✔
188	return nextBatch();	1✔
189	}
190
191	if (batchIdx >= numBatched) {	×
192	if (!nextBatch()) return false;	×
193	}
194	++batchIdx;	×
195	return true;	×
196	}
197
198	@Override
199	public Object getCurrentValue() throws IOException, InterruptedException {
200	if (returnColumnarBatch) {	1✔
201	int value = vectorProxy.numRows();	1✔
202	rowCount += value;	1✔
203	if (inputMetricsStats != null) {	1✔
204	inputMetricsStats.incrementRecordRead((long) value);	×
205	}
206	return vectorProxy.getColumnarBatch();	1✔
207	}
208	rowCount += 1;	×
209	return vectorProxy.getRow(batchIdx - 1);	×
210	}
211
212	@Override
213	public Void getCurrentKey() throws IOException, InterruptedException {
214	return null;	×
215	}
216
217	@Override
218	public float getProgress() throws IOException, InterruptedException {
219	// TODO : Implement it based on total number of rows it is going to retrieve.
220	return 0;	×
221	}
222
223	/**
224	* Returns the ColumnarBatch object that will be used for all rows returned by this reader.
225	* This object is reused. Calling this enables the vectorized reader. This should be called
226	* before any calls to nextKeyValue/nextBatch.
227	*/
228
229	public void initBatch(MemoryMode memMode, StructType partitionColumns,
230	InternalRow partitionValues) {
231	List<ProjectionDimension> queryDimension = queryModel.getProjectionDimensions();	1✔
232	List<ProjectionMeasure> queryMeasures = queryModel.getProjectionMeasures();	1✔
233	StructField[] fields = new StructField[queryDimension.size() + queryMeasures.size()];	1✔
234	this.isNoDictStringField = new boolean[queryDimension.size() + queryMeasures.size()];	1✔
235	for (int i = 0; i < queryDimension.size(); i++) {	1✔
236	ProjectionDimension dim = queryDimension.get(i);	1✔
237	if (dim.getDimension().hasEncoding(Encoding.DIRECT_DICTIONARY)) {	1✔
238	DirectDictionaryGenerator generator = DirectDictionaryKeyGeneratorFactory	1✔
239	.getDirectDictionaryGenerator(dim.getDimension().getDataType());	1✔
240	fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),	1✔
241	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(generator.getReturnType()), true, null);	1✔
242	} else if (!dim.getDimension().hasEncoding(Encoding.DICTIONARY)) {	1✔
243	if (dim.getDimension().getDataType() == DataTypes.STRING	1✔
244	\|\| dim.getDimension().getDataType() == DataTypes.VARCHAR \|\| dim.getDimension()	1✔
245	.getColumnSchema().isLocalDictColumn()) {	1✔
246	this.isNoDictStringField[dim.getOrdinal()] = true;	1✔
247	}
248	fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),	1✔
249	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(dim.getDimension().getDataType()), true,	1✔
250	null);
251	} else if (dim.getDimension().isComplex()) {	×
252	fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),	×
253	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(dim.getDimension().getDataType()), true,	×
254	null);
255	} else {
256	fields[dim.getOrdinal()] = new StructField(dim.getColumnName(),	×
257	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(DataTypes.INT), true, null);	×
258	}
259	}
260
261	for (int i = 0; i < queryMeasures.size(); i++) {	1✔
262	ProjectionMeasure msr = queryMeasures.get(i);	1✔
263	DataType dataType = msr.getMeasure().getDataType();	1✔
264	if (dataType == DataTypes.BOOLEAN \|\| dataType == DataTypes.SHORT \|\| dataType == DataTypes.INT	1✔
265	\|\| dataType == DataTypes.LONG \|\| dataType == DataTypes.FLOAT
266	\|\| dataType == DataTypes.BYTE) {
267	fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),	1✔
268	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(msr.getMeasure().getDataType()), true,	1✔
269	null);
270	} else if (DataTypes.isDecimal(dataType)) {	1✔
271	fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),	1✔
272	new DecimalType(msr.getMeasure().getPrecision(), msr.getMeasure().getScale()), true,	1✔
273	null);
274	} else {
275	fields[msr.getOrdinal()] = new StructField(msr.getColumnName(),	1✔
276	CarbonSparkDataSourceUtil.convertCarbonToSparkDataType(DataTypes.DOUBLE), true, null);	1✔
277	}
278	}
279	StructType schema = new StructType(fields);	1✔
280	if (partitionColumns != null) {	1✔
281	for (StructField field : partitionColumns.fields()) {	1✔
282	schema = schema.add(field);	1✔
283	}
284	}
285	short batchSize = DEFAULT_BATCH_SIZE;	1✔
286	if (queryModel.isDirectVectorFill()) {	1✔
287	batchSize = CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT;	1✔
288	}
289	vectorProxy = new CarbonVectorProxy(DEFAULT_MEMORY_MODE, schema, batchSize);	1✔
290
291	if (partitionColumns != null) {	1✔
292	int partitionIdx = fields.length;	1✔
293	for (int i = 0; i < partitionColumns.fields().length; i++) {	1✔
294	ColumnVectorUtils.populate(vectorProxy.column(i + partitionIdx), partitionValues, i);	1✔
295	vectorProxy.column(i + partitionIdx).setIsConstant();	1✔
296	}
297	}
298	CarbonColumnVector[] vectors = new CarbonColumnVector[fields.length];	1✔
299	boolean[] filteredRows = null;	1✔
300	if (queryModel.isDirectVectorFill()) {	1✔
301	for (int i = 0; i < fields.length; i++) {	1✔
302	vectors[i] = new ColumnarVectorWrapperDirect(vectorProxy, i);	1✔
303	if (isNoDictStringField[i]) {	1✔
304	((ColumnarVectorWrapperDirect) vectors[i]).reserveDictionaryIds();	1✔
305	}
306	}
307	} else {
308	filteredRows = new boolean[vectorProxy.numRows()];	×
309	for (int i = 0; i < fields.length; i++) {	×
310	vectors[i] = new ColumnarVectorWrapper(vectorProxy, i);	×
311	if (isNoDictStringField[i]) {	×
312	if (vectors[i] instanceof ColumnarVectorWrapper) {	×
313	((ColumnarVectorWrapper) vectors[i]).reserveDictionaryIds();	×
314	}
315	}
316	}
317	}
318	carbonColumnarBatch = new CarbonColumnarBatch(vectors, vectorProxy.numRows(), filteredRows);	1✔
319	}	1✔
320
321	private void initBatch() {
322	initBatch(DEFAULT_MEMORY_MODE, new StructType(), InternalRow.empty());	×
323	}	×
324
325	private void resultBatch() {
326	if (vectorProxy == null) initBatch();	1✔
327	}	1✔
328
329
330
331	/**
332	* Advances to the next batch of rows. Returns false if there are no more.
333	*/
334	private boolean nextBatch() {
335	if (null != isNoDictStringField) {	1✔
336	for (int i = 0; i < isNoDictStringField.length; i++) {	1✔
337	if (isNoDictStringField[i]) {	1✔
338	vectorProxy.resetDictionaryIds(i);	1✔
339	}
340	}
341	}
342	vectorProxy.reset();	1✔
343	carbonColumnarBatch.reset();	1✔
344	if (iterator.hasNext()) {	1✔
345	iterator.processNextBatch(carbonColumnarBatch);	1✔
346	int actualSize = carbonColumnarBatch.getActualSize();	1✔
347	vectorProxy.setNumRows(actualSize);	1✔
348	numBatched = actualSize;	1✔
349	batchIdx = 0;	1✔
350	return true;	1✔
351	}
352	return false;	1✔
353	}
354
355	}

apache / carbondata / 9614

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