3288

Build # 3288

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/579db7346eaad954d605bca8ed2b576643c03add">579db7346<a href="https://github.com/apache/carbondata/commit/579db7346eaad954d605bca8ed2b576643c03add">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/79b697dc0a6dd9401acd01f148c9ca3f16a14619">79b697dc0</a>">79b697dc0</a><a href="https://github.com/apache/carbondata/commit/579db7346eaad954d605bca8ed2b576643c03add"> into eb7a83350">eb7a83350</a>

Run Details

66 of 66 new or added lines in 6 files covered. (100.0%)

63461 of 80118 relevant lines covered (79.21%)

1.05 hits per line

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

93.7

/core/src/main/java/org/apache/carbondata/core/scan/result/iterator/RawResultIterator.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.result.iterator;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.carbondata.common.CarbonIterator;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.keygenerator.KeyGenException;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.encoder.Encoding;
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.executor.util.RestructureUtil;
import org.apache.carbondata.core.scan.result.RowBatch;
import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.DataTypeUtil;

import org.apache.log4j.Logger;

/**
 * This is a wrapper iterator over the detail raw query iterator.
 * This iterator will handle the processing of the raw rows.
 * This will handle the batch results and will iterate on the batches and give single row.
 */
public class RawResultIterator extends CarbonIterator<Object[]> {

  private final SegmentProperties sourceSegProperties;

  private final SegmentProperties destinationSegProperties;
  /**
   * Iterator of the Batch raw result.
   */
  private CarbonIterator<RowBatch> detailRawQueryResultIterator;

  private boolean prefetchEnabled;
  private List<Object[]> currentBuffer;
  private List<Object[]> backupBuffer;
  private int currentIdxInBuffer;
  private ExecutorService executorService;
  private Future<Void> fetchFuture;
  private Object[] currentRawRow = null;
  private boolean isBackupFilled = false;

  // column reorder for no-dictionary column
  private int noDictCount;
  private int[] noDictMap;
  // column drift
  private final boolean hasColumnDrift;
  private boolean[] isColumnDrift;
  private int measureCount;
  private DataType[] measureDataTypes;

  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(RawResultIterator.class.getName());

  public RawResultIterator(CarbonIterator<RowBatch> detailRawQueryResultIterator,
      SegmentProperties sourceSegProperties, SegmentProperties destinationSegProperties,
      boolean isStreamingHandoff, boolean hasColumnDrift) {
    this.detailRawQueryResultIterator = detailRawQueryResultIterator;
    this.sourceSegProperties = sourceSegProperties;
    this.destinationSegProperties = destinationSegProperties;
    this.executorService = Executors.newFixedThreadPool(1);
    this.hasColumnDrift = hasColumnDrift;
    if (!isStreamingHandoff) {
      init();
    }
  }

  private void initForColumnDrift() {
    List<CarbonDimension> noDictDims =
        new ArrayList<>(destinationSegProperties.getDimensions().size());
    for (CarbonDimension dimension : destinationSegProperties.getDimensions()) {
      if (dimension.getNumberOfChild() == 0) {
        if (!dimension.hasEncoding(Encoding.DICTIONARY)) {
          noDictDims.add(dimension);
        }
      }
    }
    measureCount = destinationSegProperties.getMeasures().size();
    noDictCount = noDictDims.size();
    isColumnDrift = new boolean[noDictCount];
    noDictMap = new int[noDictCount];
    measureDataTypes = new DataType[noDictCount];
    List<CarbonMeasure> sourceMeasures = sourceSegProperties.getMeasures();
    int tableMeasureCount = sourceMeasures.size();
    for (int i = 0; i < noDictCount; i++) {
      for (int j = 0; j < tableMeasureCount; j++) {
        if (RestructureUtil.isColumnMatches(true, noDictDims.get(i), sourceMeasures.get(j))) {
          isColumnDrift[i] = true;
          measureDataTypes[i] = sourceMeasures.get(j).getDataType();
          break;
        }
      }
      if (measureDataTypes[i] == null) {
        isColumnDrift[i] = false;
      }
    }
    int noDictIndex = 0;
    // the column drift are at the end of measures
    int measureIndex = measureCount + 1;
    for (int i = 0; i < noDictCount; i++) {
      if (isColumnDrift[i]) {
        noDictMap[i] = measureIndex++;
      } else {
        noDictMap[i] = noDictIndex++;
      }
    }
  }

  private void init() {
    if (hasColumnDrift) {
      initForColumnDrift();
    }
    this.prefetchEnabled = CarbonProperties.getInstance().getProperty(
        CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE,
        CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE_DEFAULT).equalsIgnoreCase("true");
    try {
      new RowsFetcher(false).call();
      if (prefetchEnabled) {
        this.fetchFuture = executorService.submit(new RowsFetcher(true));
      }
    } catch (Exception e) {
      LOGGER.error("Error occurs while fetching records", e);
      throw new RuntimeException(e);
    }
  }

  /**
   * fetch rows
   */
  private final class RowsFetcher implements Callable<Void> {
    private boolean isBackupFilling;

    private RowsFetcher(boolean isBackupFilling) {
      this.isBackupFilling = isBackupFilling;
    }

    @Override
    public Void call() throws Exception {
      if (isBackupFilling) {
        backupBuffer = fetchRows();
        isBackupFilled = true;
      } else {
        currentBuffer = fetchRows();
      }
      return null;
    }
  }

  private List<Object[]> fetchRows() throws Exception {
    List<Object[]> converted = new ArrayList<>();
    if (detailRawQueryResultIterator.hasNext()) {
      for (Object[] r : detailRawQueryResultIterator.next().getRows()) {
        converted.add(convertRow(r));
      }
    }
    return converted;
  }

  private void fillDataFromPrefetch() {
    try {
      if (currentIdxInBuffer >= currentBuffer.size() && 0 != currentIdxInBuffer) {
        if (prefetchEnabled) {
          if (!isBackupFilled) {
            fetchFuture.get();
          }
          // copy backup buffer to current buffer and fill backup buffer asyn
          currentIdxInBuffer = 0;
          currentBuffer.clear();
          currentBuffer = backupBuffer;
          isBackupFilled = false;
          fetchFuture = executorService.submit(new RowsFetcher(true));
        } else {
          currentIdxInBuffer = 0;
          new RowsFetcher(false).call();
        }
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * populate a row with index counter increased
   */
  private void popRow() {
    fillDataFromPrefetch();
    currentRawRow = currentBuffer.get(currentIdxInBuffer);
    currentIdxInBuffer++;
  }

  /**
   * populate a row with index counter unchanged
   */
  private void pickRow() {
    fillDataFromPrefetch();
    currentRawRow = currentBuffer.get(currentIdxInBuffer);
  }

  @Override
  public boolean hasNext() {
    fillDataFromPrefetch();
    if (currentIdxInBuffer < currentBuffer.size()) {
      return true;
    }

    return false;
  }

  @Override
  public Object[] next() {
    try {
      popRow();
      return this.currentRawRow;
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * for fetching the row with out incrementing counter.
   * @return
   */
  public Object[] fetchConverted() throws KeyGenException {
    pickRow();
    return this.currentRawRow;
  }

  private Object[] convertRow(Object[] rawRow) throws KeyGenException {
    ByteArrayWrapper dimObject = (ByteArrayWrapper) rawRow[0];
    byte[] dims = dimObject.getDictionaryKey();
    long[] keyArray = sourceSegProperties.getDimensionKeyGenerator().getKeyArray(dims);
    byte[] covertedBytes =
        destinationSegProperties.getDimensionKeyGenerator().generateKey(keyArray);
    dimObject.setDictionaryKey(covertedBytes);
    if (hasColumnDrift) {
      // need move measure to dimension and return new row by current schema
      byte[][] noDicts = dimObject.getNoDictionaryKeys();
      byte[][] newNoDicts = new byte[noDictCount][];
      for (int i = 0; i < noDictCount; i++) {
        if (isColumnDrift[i]) {
          newNoDicts[i] = DataTypeUtil
              .getBytesDataDataTypeForNoDictionaryColumn(rawRow[noDictMap[i]], measureDataTypes[i]);
        } else {
          newNoDicts[i] = noDicts[noDictMap[i]];
        }
      }
      ByteArrayWrapper newWrapper = new ByteArrayWrapper();
      newWrapper.setDictionaryKey(covertedBytes);
      newWrapper.setNoDictionaryKeys(newNoDicts);
      newWrapper.setComplexTypesKeys(dimObject.getComplexTypesKeys());
      newWrapper.setImplicitColumnByteArray(dimObject.getImplicitColumnByteArray());
      Object[] finalRawRow = new Object[1 + measureCount];
      finalRawRow[0] = newWrapper;
      System.arraycopy(rawRow, 1, finalRawRow, 1, measureCount);
      return finalRawRow;
    }
    return rawRow;
  }

  public void close() {
    if (null != executorService) {
      executorService.shutdownNow();
    }
  }
}

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.result.iterator;
18
19	import java.util.ArrayList;
20	import java.util.List;
21	import java.util.concurrent.Callable;
22	import java.util.concurrent.ExecutorService;
23	import java.util.concurrent.Executors;
24	import java.util.concurrent.Future;
25
26	import org.apache.carbondata.common.CarbonIterator;
27	import org.apache.carbondata.common.logging.LogServiceFactory;
28	import org.apache.carbondata.core.constants.CarbonCommonConstants;
29	import org.apache.carbondata.core.datastore.block.SegmentProperties;
30	import org.apache.carbondata.core.keygenerator.KeyGenException;
31	import org.apache.carbondata.core.metadata.datatype.DataType;
32	import org.apache.carbondata.core.metadata.encoder.Encoding;
33	import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
34	import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
35	import org.apache.carbondata.core.scan.executor.util.RestructureUtil;
36	import org.apache.carbondata.core.scan.result.RowBatch;
37	import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
38	import org.apache.carbondata.core.util.CarbonProperties;
39	import org.apache.carbondata.core.util.DataTypeUtil;
40
41	import org.apache.log4j.Logger;
42
43	/**
44	* This is a wrapper iterator over the detail raw query iterator.
45	* This iterator will handle the processing of the raw rows.
46	* This will handle the batch results and will iterate on the batches and give single row.
47	*/
48	public class RawResultIterator extends CarbonIterator<Object[]> {
49
50	private final SegmentProperties sourceSegProperties;
51
52	private final SegmentProperties destinationSegProperties;
53	/**
54	* Iterator of the Batch raw result.
55	*/
56	private CarbonIterator<RowBatch> detailRawQueryResultIterator;
57
58	private boolean prefetchEnabled;
59	private List<Object[]> currentBuffer;
60	private List<Object[]> backupBuffer;
61	private int currentIdxInBuffer;
62	private ExecutorService executorService;
63	private Future<Void> fetchFuture;
64	private Object[] currentRawRow = null;	1✔
65	private boolean isBackupFilled = false;	1✔
66
67	// column reorder for no-dictionary column
68	private int noDictCount;
69	private int[] noDictMap;
70	// column drift
71	private final boolean hasColumnDrift;
72	private boolean[] isColumnDrift;
73	private int measureCount;
74	private DataType[] measureDataTypes;
75
76	/**
77	* LOGGER
78	*/
79	private static final Logger LOGGER =	1✔
80	LogServiceFactory.getLogService(RawResultIterator.class.getName());	1✔
81
82	public RawResultIterator(CarbonIterator<RowBatch> detailRawQueryResultIterator,
83	SegmentProperties sourceSegProperties, SegmentProperties destinationSegProperties,
84	boolean isStreamingHandoff, boolean hasColumnDrift) {	1✔
85	this.detailRawQueryResultIterator = detailRawQueryResultIterator;	1✔
86	this.sourceSegProperties = sourceSegProperties;	1✔
87	this.destinationSegProperties = destinationSegProperties;	1✔
88	this.executorService = Executors.newFixedThreadPool(1);	1✔
89	this.hasColumnDrift = hasColumnDrift;	1✔
90	if (!isStreamingHandoff) {	1✔
91	init();	1✔
92	}
93	}	1✔
94
95	private void initForColumnDrift() {
96	List<CarbonDimension> noDictDims =	1✔
97	new ArrayList<>(destinationSegProperties.getDimensions().size());	1✔
98	for (CarbonDimension dimension : destinationSegProperties.getDimensions()) {	1✔
99	if (dimension.getNumberOfChild() == 0) {	1✔
100	if (!dimension.hasEncoding(Encoding.DICTIONARY)) {	1✔
101	noDictDims.add(dimension);	1✔
102	}
103	}
104	}	1✔
105	measureCount = destinationSegProperties.getMeasures().size();	1✔
106	noDictCount = noDictDims.size();	1✔
107	isColumnDrift = new boolean[noDictCount];	1✔
108	noDictMap = new int[noDictCount];	1✔
109	measureDataTypes = new DataType[noDictCount];	1✔
110	List<CarbonMeasure> sourceMeasures = sourceSegProperties.getMeasures();	1✔
111	int tableMeasureCount = sourceMeasures.size();	1✔
112	for (int i = 0; i < noDictCount; i++) {	1✔
113	for (int j = 0; j < tableMeasureCount; j++) {	1✔
114	if (RestructureUtil.isColumnMatches(true, noDictDims.get(i), sourceMeasures.get(j))) {	1✔
115	isColumnDrift[i] = true;	1✔
116	measureDataTypes[i] = sourceMeasures.get(j).getDataType();	1✔
117	break;	1✔
118	}
119	}
120	if (measureDataTypes[i] == null) {	1✔
121	isColumnDrift[i] = false;	1✔
122	}
123	}
124	int noDictIndex = 0;	1✔
125	// the column drift are at the end of measures
126	int measureIndex = measureCount + 1;	1✔
127	for (int i = 0; i < noDictCount; i++) {	1✔
128	if (isColumnDrift[i]) {	1✔
129	noDictMap[i] = measureIndex++;	1✔
130	} else {
131	noDictMap[i] = noDictIndex++;	1✔
132	}
133	}
134	}	1✔
135
136	private void init() {
137	if (hasColumnDrift) {	1✔
138	initForColumnDrift();	1✔
139	}
140	this.prefetchEnabled = CarbonProperties.getInstance().getProperty(	1✔
141	CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE,
142	CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE_DEFAULT).equalsIgnoreCase("true");	1✔
143	try {
144	new RowsFetcher(false).call();	1✔
145	if (prefetchEnabled) {	1✔
146	this.fetchFuture = executorService.submit(new RowsFetcher(true));	1✔
147	}
148	} catch (Exception e) {	×
149	LOGGER.error("Error occurs while fetching records", e);	×
150	throw new RuntimeException(e);	×
151	}	1✔
152	}	1✔
153
154	/**
155	* fetch rows
156	*/
157	private final class RowsFetcher implements Callable<Void> {
158	private boolean isBackupFilling;
159
160	private RowsFetcher(boolean isBackupFilling) {	1✔
161	this.isBackupFilling = isBackupFilling;	1✔
162	}	1✔
163
164	@Override
165	public Void call() throws Exception {
166	if (isBackupFilling) {	1✔
167	backupBuffer = fetchRows();	1✔
168	isBackupFilled = true;	1✔
169	} else {
170	currentBuffer = fetchRows();	1✔
171	}
172	return null;	1✔
173	}
174	}
175
176	private List<Object[]> fetchRows() throws Exception {
177	List<Object[]> converted = new ArrayList<>();	1✔
178	if (detailRawQueryResultIterator.hasNext()) {	1✔
179	for (Object[] r : detailRawQueryResultIterator.next().getRows()) {	1✔
180	converted.add(convertRow(r));	1✔
181	}	1✔
182	}
183	return converted;	1✔
184	}
185
186	private void fillDataFromPrefetch() {
187	try {
188	if (currentIdxInBuffer >= currentBuffer.size() && 0 != currentIdxInBuffer) {	1✔
189	if (prefetchEnabled) {	1✔
190	if (!isBackupFilled) {	1✔
191	fetchFuture.get();	×
192	}
193	// copy backup buffer to current buffer and fill backup buffer asyn
194	currentIdxInBuffer = 0;	1✔
195	currentBuffer.clear();	1✔
196	currentBuffer = backupBuffer;	1✔
197	isBackupFilled = false;	1✔
198	fetchFuture = executorService.submit(new RowsFetcher(true));	1✔
199	} else {
200	currentIdxInBuffer = 0;	1✔
201	new RowsFetcher(false).call();	1✔
202	}
203	}
204	} catch (Exception e) {	×
205	throw new RuntimeException(e);	×
206	}	1✔
207	}	1✔
208
209	/**
210	* populate a row with index counter increased
211	*/
212	private void popRow() {
213	fillDataFromPrefetch();	1✔
214	currentRawRow = currentBuffer.get(currentIdxInBuffer);	1✔
215	currentIdxInBuffer++;	1✔
216	}	1✔
217
218	/**
219	* populate a row with index counter unchanged
220	*/
221	private void pickRow() {
222	fillDataFromPrefetch();	1✔
223	currentRawRow = currentBuffer.get(currentIdxInBuffer);	1✔
224	}	1✔
225
226	@Override
227	public boolean hasNext() {
228	fillDataFromPrefetch();	1✔
229	if (currentIdxInBuffer < currentBuffer.size()) {	1✔
230	return true;	1✔
231	}
232
233	return false;	1✔
234	}
235
236	@Override
237	public Object[] next() {
238	try {
239	popRow();	1✔
240	return this.currentRawRow;	1✔
241	} catch (Exception e) {	×
242	throw new RuntimeException(e);	×
243	}
244	}
245
246	/**
247	* for fetching the row with out incrementing counter.
248	* @return
249	*/
250	public Object[] fetchConverted() throws KeyGenException {
251	pickRow();	1✔
252	return this.currentRawRow;	1✔
253	}
254
255	private Object[] convertRow(Object[] rawRow) throws KeyGenException {
256	ByteArrayWrapper dimObject = (ByteArrayWrapper) rawRow[0];	1✔
257	byte[] dims = dimObject.getDictionaryKey();	1✔
258	long[] keyArray = sourceSegProperties.getDimensionKeyGenerator().getKeyArray(dims);	1✔
259	byte[] covertedBytes =	1✔
260	destinationSegProperties.getDimensionKeyGenerator().generateKey(keyArray);	1✔
261	dimObject.setDictionaryKey(covertedBytes);	1✔
262	if (hasColumnDrift) {	1✔
263	// need move measure to dimension and return new row by current schema
264	byte[][] noDicts = dimObject.getNoDictionaryKeys();	1✔
265	byte[][] newNoDicts = new byte[noDictCount][];	1✔
266	for (int i = 0; i < noDictCount; i++) {	1✔
267	if (isColumnDrift[i]) {	1✔
268	newNoDicts[i] = DataTypeUtil	1✔
269	.getBytesDataDataTypeForNoDictionaryColumn(rawRow[noDictMap[i]], measureDataTypes[i]);	1✔
270	} else {
271	newNoDicts[i] = noDicts[noDictMap[i]];	1✔
272	}
273	}
274	ByteArrayWrapper newWrapper = new ByteArrayWrapper();	1✔
275	newWrapper.setDictionaryKey(covertedBytes);	1✔
276	newWrapper.setNoDictionaryKeys(newNoDicts);	1✔
277	newWrapper.setComplexTypesKeys(dimObject.getComplexTypesKeys());	1✔
278	newWrapper.setImplicitColumnByteArray(dimObject.getImplicitColumnByteArray());	1✔
279	Object[] finalRawRow = new Object[1 + measureCount];	1✔
280	finalRawRow[0] = newWrapper;	1✔
281	System.arraycopy(rawRow, 1, finalRawRow, 1, measureCount);	1✔
282	return finalRawRow;	1✔
283	}
284	return rawRow;	1✔
285	}
286
287	public void close() {
288	if (null != executorService) {	1✔
289	executorService.shutdownNow();	1✔
290	}
291	}	1✔
292	}

apache / carbondata / 3288

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