apache / carbondata / 598

Build # 598

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/3f72ac2e6ef1a8c0537feb9de826ccfd858d1513">3f72ac2e6">&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/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/3f72ac2e6ef1a8c0537feb9de826ccfd858d1513">3f72ac2e6</a><a href="https://github.com/apache/carbondata/commit/3f72ac2e6ef1a8c0537feb9de826ccfd858d1513">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/3f72ac2e6ef1a8c0537feb9de826ccfd858d1513&quot;&gt;3f72ac2e6&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/3f72ac2e6ef1a8c0537feb9de826ccfd858d1513&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 ... (continued)

Run Details

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

59748 of 74251 relevant lines covered (80.47%)

1.07 hits per line

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

98.06

/core/src/main/java/org/apache/carbondata/core/scan/processor/DataBlockIterator.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.core.scan.processor;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.carbondata.common.CarbonIterator;
import org.apache.carbondata.core.datastore.DataRefNode;
import org.apache.carbondata.core.datastore.FileReader;
import org.apache.carbondata.core.scan.collector.ResultCollectorFactory;
import org.apache.carbondata.core.scan.collector.ScannedResultCollector;
import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.core.scan.result.BlockletScannedResult;
import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
import org.apache.carbondata.core.scan.scanner.BlockletScanner;
import org.apache.carbondata.core.scan.scanner.impl.BlockletFilterScanner;
import org.apache.carbondata.core.scan.scanner.impl.BlockletFullScanner;
import org.apache.carbondata.core.stats.QueryStatisticsModel;
import org.apache.carbondata.core.util.TaskMetricsMap;

/**
 * This abstract class provides a skeletal implementation of the
 * Block iterator.
 */
public class DataBlockIterator extends CarbonIterator<List<Object[]>> {

  /**
   * iterator which will be used to iterate over blocklets
   */
  private BlockletIterator blockletIterator;

  /**
   * result collector which will be used to aggregate the scanned result
   */
  private ScannedResultCollector scannerResultAggregator;

  /**
   * processor which will be used to process the block processing can be
   * filter processing or non filter processing
   */
  private BlockletScanner blockletScanner;

  /**
   * batch size of result
   */
  private int batchSize;

  private ExecutorService executorService;

  private Future<BlockletScannedResult> future;

  private Future<RawBlockletColumnChunks> futureIo;

  private BlockletScannedResult scannedResult;

  private BlockExecutionInfo blockExecutionInfo;

  private FileReader fileReader;

  private AtomicBoolean nextBlock;

  private AtomicBoolean nextRead;

  public DataBlockIterator(BlockExecutionInfo blockExecutionInfo, FileReader fileReader,
      int batchSize, QueryStatisticsModel queryStatisticsModel, ExecutorService executorService) {
    this.blockExecutionInfo = blockExecutionInfo;
    this.blockExecutionInfo.setQueryStatisticsModel(queryStatisticsModel);
    this.fileReader = fileReader;
    blockletIterator = new BlockletIterator(blockExecutionInfo.getFirstDataBlock(),
        blockExecutionInfo.getNumberOfBlockToScan());
    if (blockExecutionInfo.getFilterExecuterTree() != null) {
      blockletScanner = new BlockletFilterScanner(blockExecutionInfo, queryStatisticsModel);
    } else {
      blockletScanner = new BlockletFullScanner(blockExecutionInfo, queryStatisticsModel);
    }
    this.scannerResultAggregator =
        ResultCollectorFactory.getScannedResultCollector(blockExecutionInfo);
    this.batchSize = batchSize;
    this.executorService = executorService;
    this.nextBlock = new AtomicBoolean(false);
    this.nextRead = new AtomicBoolean(false);
  }

  @Override
  public List<Object[]> next() {
    List<Object[]> collectedResult = null;
    if (updateScanner()) {
      collectedResult = this.scannerResultAggregator.collectResultInRow(scannedResult, batchSize);
      while (collectedResult.size() < batchSize && updateScanner()) {
        List<Object[]> data = this.scannerResultAggregator
            .collectResultInRow(scannedResult, batchSize - collectedResult.size());
        collectedResult.addAll(data);
      }
    } else {
      collectedResult = new ArrayList<>();
    }
    return collectedResult;
  }

  @Override
  public boolean hasNext() {
    if (scannedResult != null && scannedResult.hasNext()) {
      return true;
    } else {
      if (null != scannedResult) {
        scannedResult.freeMemory();
      }
      return blockletIterator.hasNext() || nextBlock.get() || nextRead.get();
    }
  }

  /**
   * Return true if scan result if non-empty
   */
  private boolean updateScanner() {
    try {
      if (scannedResult != null && scannedResult.hasNext()) {
        return true;
      } else {
        scannedResult = processNextBlocklet();
        while (scannedResult != null) {
          if (scannedResult.hasNext()) {
            return true;
          }
          scannedResult = processNextBlocklet();
        }
        nextBlock.set(false);
        nextRead.set(false);
        return false;
      }
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }

  private BlockletScannedResult processNextBlocklet() throws Exception {
    BlockletScannedResult result = null;
    if (blockExecutionInfo.isPrefetchBlocklet()) {
      if (blockletIterator.hasNext() || nextBlock.get() || nextRead.get()) {
        if (future == null) {
          future = scanNextBlockletAsync();
        }
        result = future.get();
        nextBlock.set(false);
        if (blockletIterator.hasNext() || nextRead.get()) {
          nextBlock.set(true);
          future = scanNextBlockletAsync();
        }
      }
    } else {
      if (blockletIterator.hasNext()) {
        RawBlockletColumnChunks rawChunks = readNextBlockletColumnChunks();
        if (rawChunks != null) {
          result = blockletScanner.scanBlocklet(rawChunks);
        }
      }
    }
    return result;
  }

  private RawBlockletColumnChunks readNextBlockletColumnChunks() throws IOException {
    RawBlockletColumnChunks rawBlockletColumnChunks = getNextBlockletColumnChunks();
    if (rawBlockletColumnChunks != null) {
      blockletScanner.readBlocklet(rawBlockletColumnChunks);
      return rawBlockletColumnChunks;
    }
    return null;
  }

  private RawBlockletColumnChunks getNextBlockletColumnChunks() {
    RawBlockletColumnChunks rawBlockletColumnChunks = null;
    do {
      DataRefNode dataBlock = blockletIterator.next();
      if (dataBlock.getColumnsMaxValue() == null || blockletScanner.isScanRequired(dataBlock)) {
        rawBlockletColumnChunks =  RawBlockletColumnChunks.newInstance(
            blockExecutionInfo.getTotalNumberDimensionToRead(),
            blockExecutionInfo.getTotalNumberOfMeasureToRead(), fileReader, dataBlock);
      }
    } while (rawBlockletColumnChunks == null && blockletIterator.hasNext());
    return rawBlockletColumnChunks;
  }

  private Future<BlockletScannedResult> scanNextBlockletAsync() {
    return executorService.submit(new Callable<BlockletScannedResult>() {
      @Override public BlockletScannedResult call() throws Exception {
        if (futureIo == null) {
          futureIo = readNextBlockletAsync();
        }
        RawBlockletColumnChunks rawBlockletColumnChunks = futureIo.get();
        futureIo = null;
        nextRead.set(false);
        if (rawBlockletColumnChunks != null) {
          if (blockletIterator.hasNext()) {
            nextRead.set(true);
            futureIo = readNextBlockletAsync();
          }
          return blockletScanner.scanBlocklet(rawBlockletColumnChunks);
        }
        return null;
      }
    });
  }

  private Future<RawBlockletColumnChunks> readNextBlockletAsync() {
    return executorService.submit(new Callable<RawBlockletColumnChunks>() {
      @Override public RawBlockletColumnChunks call() throws Exception {
        try {
          TaskMetricsMap.getInstance().registerThreadCallback();
          if (blockletIterator.hasNext()) {
            return readNextBlockletColumnChunks();
          } else {
            return null;
          }
        } finally {
          // update read bytes metrics for this thread
          TaskMetricsMap.getInstance().updateReadBytes(Thread.currentThread().getId());
        }
      }
    });
  }

  public void processNextBatch(CarbonColumnarBatch columnarBatch) {
    if (updateScanner()) {
      this.scannerResultAggregator.collectResultInColumnarBatch(scannedResult, columnarBatch);
    }
  }


  /**
   * Close the resources
   */
  public void close() {
    // free the current scanned result
    if (null != scannedResult && !scannedResult.hasNext()) {
      scannedResult.freeMemory();
    }
    // free any pre-fetched memory if present
    if (null != future) {
      try {
        BlockletScannedResult blockletScannedResult = future.get();
        if (blockletScannedResult != null) {
          blockletScannedResult.freeMemory();
        }
      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(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	package org.apache.carbondata.core.scan.processor;
18
19	import java.io.IOException;
20	import java.util.ArrayList;
21	import java.util.List;
22	import java.util.concurrent.Callable;
23	import java.util.concurrent.ExecutionException;
24	import java.util.concurrent.ExecutorService;
25	import java.util.concurrent.Future;
26	import java.util.concurrent.atomic.AtomicBoolean;
27
28	import org.apache.carbondata.common.CarbonIterator;
29	import org.apache.carbondata.core.datastore.DataRefNode;
30	import org.apache.carbondata.core.datastore.FileReader;
31	import org.apache.carbondata.core.scan.collector.ResultCollectorFactory;
32	import org.apache.carbondata.core.scan.collector.ScannedResultCollector;
33	import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
34	import org.apache.carbondata.core.scan.result.BlockletScannedResult;
35	import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
36	import org.apache.carbondata.core.scan.scanner.BlockletScanner;
37	import org.apache.carbondata.core.scan.scanner.impl.BlockletFilterScanner;
38	import org.apache.carbondata.core.scan.scanner.impl.BlockletFullScanner;
39	import org.apache.carbondata.core.stats.QueryStatisticsModel;
40	import org.apache.carbondata.core.util.TaskMetricsMap;
41
42	/**
43	* This abstract class provides a skeletal implementation of the
44	* Block iterator.
45	*/
46	public class DataBlockIterator extends CarbonIterator<List<Object[]>> {
47
48	/**
49	* iterator which will be used to iterate over blocklets
50	*/
51	private BlockletIterator blockletIterator;
52
53	/**
54	* result collector which will be used to aggregate the scanned result
55	*/
56	private ScannedResultCollector scannerResultAggregator;
57
58	/**
59	* processor which will be used to process the block processing can be
60	* filter processing or non filter processing
61	*/
62	private BlockletScanner blockletScanner;
63
64	/**
65	* batch size of result
66	*/
67	private int batchSize;
68
69	private ExecutorService executorService;
70
71	private Future<BlockletScannedResult> future;
72
73	private Future<RawBlockletColumnChunks> futureIo;
74
75	private BlockletScannedResult scannedResult;
76
77	private BlockExecutionInfo blockExecutionInfo;
78
79	private FileReader fileReader;
80
81	private AtomicBoolean nextBlock;
82
83	private AtomicBoolean nextRead;
84
85	public DataBlockIterator(BlockExecutionInfo blockExecutionInfo, FileReader fileReader,
86	int batchSize, QueryStatisticsModel queryStatisticsModel, ExecutorService executorService) {	1✔
87	this.blockExecutionInfo = blockExecutionInfo;	1✔
88	this.blockExecutionInfo.setQueryStatisticsModel(queryStatisticsModel);	1✔
89	this.fileReader = fileReader;	1✔
90	blockletIterator = new BlockletIterator(blockExecutionInfo.getFirstDataBlock(),	1✔
91	blockExecutionInfo.getNumberOfBlockToScan());	1✔
92	if (blockExecutionInfo.getFilterExecuterTree() != null) {	1✔
93	blockletScanner = new BlockletFilterScanner(blockExecutionInfo, queryStatisticsModel);	1✔
94	} else {
95	blockletScanner = new BlockletFullScanner(blockExecutionInfo, queryStatisticsModel);	1✔
96	}
97	this.scannerResultAggregator =	1✔
98	ResultCollectorFactory.getScannedResultCollector(blockExecutionInfo);	1✔
99	this.batchSize = batchSize;	1✔
100	this.executorService = executorService;	1✔
101	this.nextBlock = new AtomicBoolean(false);	1✔
102	this.nextRead = new AtomicBoolean(false);	1✔
103	}	1✔
104
105	@Override
106	public List<Object[]> next() {
107	List<Object[]> collectedResult = null;	1✔
108	if (updateScanner()) {	1✔
109	collectedResult = this.scannerResultAggregator.collectResultInRow(scannedResult, batchSize);	1✔
110	while (collectedResult.size() < batchSize && updateScanner()) {	1✔
111	List<Object[]> data = this.scannerResultAggregator	1✔
112	.collectResultInRow(scannedResult, batchSize - collectedResult.size());	1✔
113	collectedResult.addAll(data);	1✔
114	}	1✔
115	} else {
116	collectedResult = new ArrayList<>();	1✔
117	}
118	return collectedResult;	1✔
119	}
120
121	@Override
122	public boolean hasNext() {
123	if (scannedResult != null && scannedResult.hasNext()) {	1✔
124	return true;	1✔
125	} else {
126	if (null != scannedResult) {	1✔
127	scannedResult.freeMemory();	1✔
128	}
129	return blockletIterator.hasNext() \|\| nextBlock.get() \|\| nextRead.get();	1✔
130	}
131	}
132
133	/**
134	* Return true if scan result if non-empty
135	*/
136	private boolean updateScanner() {
137	try {
138	if (scannedResult != null && scannedResult.hasNext()) {	1✔
139	return true;	1✔
140	} else {
141	scannedResult = processNextBlocklet();	1✔
142	while (scannedResult != null) {	1✔
143	if (scannedResult.hasNext()) {	1✔
144	return true;	1✔
145	}
146	scannedResult = processNextBlocklet();	1✔
147	}
148	nextBlock.set(false);	1✔
149	nextRead.set(false);	1✔
150	return false;	1✔
151	}
152	} catch (Exception ex) {	1✔
153	throw new RuntimeException(ex);	1✔
154	}
155	}
156
157	private BlockletScannedResult processNextBlocklet() throws Exception {
158	BlockletScannedResult result = null;	1✔
159	if (blockExecutionInfo.isPrefetchBlocklet()) {	1✔
160	if (blockletIterator.hasNext() \|\| nextBlock.get() \|\| nextRead.get()) {	1✔
161	if (future == null) {	1✔
162	future = scanNextBlockletAsync();	1✔
163	}
164	result = future.get();	1✔
165	nextBlock.set(false);	1✔
166	if (blockletIterator.hasNext() \|\| nextRead.get()) {	1✔
167	nextBlock.set(true);	1✔
168	future = scanNextBlockletAsync();	1✔
169	}
170	}
171	} else {
172	if (blockletIterator.hasNext()) {	1✔
173	RawBlockletColumnChunks rawChunks = readNextBlockletColumnChunks();	1✔
174	if (rawChunks != null) {	1✔
175	result = blockletScanner.scanBlocklet(rawChunks);	1✔
176	}
177	}
178	}
179	return result;	1✔
180	}
181
182	private RawBlockletColumnChunks readNextBlockletColumnChunks() throws IOException {
183	RawBlockletColumnChunks rawBlockletColumnChunks = getNextBlockletColumnChunks();	1✔
184	if (rawBlockletColumnChunks != null) {	1✔
185	blockletScanner.readBlocklet(rawBlockletColumnChunks);	1✔
186	return rawBlockletColumnChunks;	1✔
187	}
188	return null;	1✔
189	}
190
191	private RawBlockletColumnChunks getNextBlockletColumnChunks() {
192	RawBlockletColumnChunks rawBlockletColumnChunks = null;	1✔
193	do {
194	DataRefNode dataBlock = blockletIterator.next();	1✔
195	if (dataBlock.getColumnsMaxValue() == null \|\| blockletScanner.isScanRequired(dataBlock)) {	1✔
196	rawBlockletColumnChunks = RawBlockletColumnChunks.newInstance(	1✔
197	blockExecutionInfo.getTotalNumberDimensionToRead(),	1✔
198	blockExecutionInfo.getTotalNumberOfMeasureToRead(), fileReader, dataBlock);	1✔
199	}
200	} while (rawBlockletColumnChunks == null && blockletIterator.hasNext());	1✔
201	return rawBlockletColumnChunks;	1✔
202	}
203
204	private Future<BlockletScannedResult> scanNextBlockletAsync() {
205	return executorService.submit(new Callable<BlockletScannedResult>() {	1✔
206	@Override public BlockletScannedResult call() throws Exception {
207	if (futureIo == null) {	1✔
208	futureIo = readNextBlockletAsync();	1✔
209	}
210	RawBlockletColumnChunks rawBlockletColumnChunks = futureIo.get();	1✔
211	futureIo = null;	1✔
212	nextRead.set(false);	1✔
213	if (rawBlockletColumnChunks != null) {	1✔
214	if (blockletIterator.hasNext()) {	1✔
215	nextRead.set(true);	1✔
216	futureIo = readNextBlockletAsync();	1✔
217	}
218	return blockletScanner.scanBlocklet(rawBlockletColumnChunks);	1✔
219	}
220	return null;	1✔
221	}
222	});
223	}
224
225	private Future<RawBlockletColumnChunks> readNextBlockletAsync() {
226	return executorService.submit(new Callable<RawBlockletColumnChunks>() {	1✔
227	@Override public RawBlockletColumnChunks call() throws Exception {
228	try {
229	TaskMetricsMap.getInstance().registerThreadCallback();	1✔
230	if (blockletIterator.hasNext()) {	1✔
231	return readNextBlockletColumnChunks();	1✔
232	} else {
233	return null;	×
234	}
235	} finally {
236	// update read bytes metrics for this thread
237	TaskMetricsMap.getInstance().updateReadBytes(Thread.currentThread().getId());	1✔
238	}	×
239	}
240	});
241	}
242
243	public void processNextBatch(CarbonColumnarBatch columnarBatch) {
244	if (updateScanner()) {	1✔
245	this.scannerResultAggregator.collectResultInColumnarBatch(scannedResult, columnarBatch);	1✔
246	}
247	}	1✔
248
249
250	/**
251	* Close the resources
252	*/
253	public void close() {
254	// free the current scanned result
255	if (null != scannedResult && !scannedResult.hasNext()) {	1✔
256	scannedResult.freeMemory();	1✔
257	}
258	// free any pre-fetched memory if present
259	if (null != future) {	1✔
260	try {
261	BlockletScannedResult blockletScannedResult = future.get();	1✔
262	if (blockletScannedResult != null) {	1✔
263	blockletScannedResult.freeMemory();	1✔
264	}
265	} catch (InterruptedException \| ExecutionException e) {	1✔
266	throw new RuntimeException(e);	1✔
267	}	1✔
268	}
269	}	1✔
270	}

apache / carbondata / 598

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