2894

Build # 2894

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/af0c648af281bde827463ed79190a0b6c9665a70">af0c648af<a href="https://github.com/apache/carbondata/commit/af0c648af281bde827463ed79190a0b6c9665a70">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/897c0d5b581d7a127d94e6f9d596728bf106c38f">897c0d5b5</a>">897c0d5b5</a><a href="https://github.com/apache/carbondata/commit/af0c648af281bde827463ed79190a0b6c9665a70"> into 3f98c5123">3f98c5123</a>

Run Details

9 of 9 new or added lines in 1 file covered. (100.0%)

62761 of 79247 relevant lines covered (79.2%)

1.05 hits per line

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

89.61

/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.scan.result.RowBatch;
import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
import org.apache.carbondata.core.util.CarbonProperties;

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;

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

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

    if (!isStreamingHandoff) {
      init();
    }
  }

  private void init() {
    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 {
    byte[] dims = ((ByteArrayWrapper) rawRow[0]).getDictionaryKey();
    long[] keyArray = sourceSegProperties.getDimensionKeyGenerator().getKeyArray(dims);
    byte[] covertedBytes =
        destinationSegProperties.getDimensionKeyGenerator().generateKey(keyArray);
    ((ByteArrayWrapper) rawRow[0]).setDictionaryKey(covertedBytes);
    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.scan.result.RowBatch;
32	import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
33	import org.apache.carbondata.core.util.CarbonProperties;
34
35	import org.apache.log4j.Logger;
36
37	/**
38	* This is a wrapper iterator over the detail raw query iterator.
39	* This iterator will handle the processing of the raw rows.
40	* This will handle the batch results and will iterate on the batches and give single row.
41	*/
42	public class RawResultIterator extends CarbonIterator<Object[]> {
43
44	private final SegmentProperties sourceSegProperties;
45
46	private final SegmentProperties destinationSegProperties;
47	/**
48	* Iterator of the Batch raw result.
49	*/
50	private CarbonIterator<RowBatch> detailRawQueryResultIterator;
51
52	private boolean prefetchEnabled;
53	private List<Object[]> currentBuffer;
54	private List<Object[]> backupBuffer;
55	private int currentIdxInBuffer;
56	private ExecutorService executorService;
57	private Future<Void> fetchFuture;
58	private Object[] currentRawRow = null;	1✔
59	private boolean isBackupFilled = false;	1✔
60
61	/**
62	* LOGGER
63	*/
64	private static final Logger LOGGER =	1✔
65	LogServiceFactory.getLogService(RawResultIterator.class.getName());	1✔
66
67	public RawResultIterator(CarbonIterator<RowBatch> detailRawQueryResultIterator,
68	SegmentProperties sourceSegProperties, SegmentProperties destinationSegProperties,
69	boolean isStreamingHandoff) {	1✔
70	this.detailRawQueryResultIterator = detailRawQueryResultIterator;	1✔
71	this.sourceSegProperties = sourceSegProperties;	1✔
72	this.destinationSegProperties = destinationSegProperties;	1✔
73	this.executorService = Executors.newFixedThreadPool(1);	1✔
74
75	if (!isStreamingHandoff) {	1✔
76	init();	1✔
77	}
78	}	1✔
79
80	private void init() {
81	this.prefetchEnabled = CarbonProperties.getInstance().getProperty(	1✔
82	CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE,
83	CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE_DEFAULT).equalsIgnoreCase("true");	1✔
84	try {
85	new RowsFetcher(false).call();	1✔
86	if (prefetchEnabled) {	1✔
87	this.fetchFuture = executorService.submit(new RowsFetcher(true));	1✔
88	}
89	} catch (Exception e) {	×
90	LOGGER.error("Error occurs while fetching records", e);	×
91	throw new RuntimeException(e);	×
92	}	1✔
93	}	1✔
94
95	/**
96	* fetch rows
97	*/
98	private final class RowsFetcher implements Callable<Void> {
99	private boolean isBackupFilling;
100
101	private RowsFetcher(boolean isBackupFilling) {	1✔
102	this.isBackupFilling = isBackupFilling;	1✔
103	}	1✔
104
105	@Override
106	public Void call() throws Exception {
107	if (isBackupFilling) {	1✔
108	backupBuffer = fetchRows();	1✔
109	isBackupFilled = true;	1✔
110	} else {
111	currentBuffer = fetchRows();	1✔
112	}
113	return null;	1✔
114	}
115	}
116
117	private List<Object[]> fetchRows() throws Exception {
118	List<Object[]> converted = new ArrayList<>();	1✔
119	if (detailRawQueryResultIterator.hasNext()) {	1✔
120	for (Object[] r : detailRawQueryResultIterator.next().getRows()) {	1✔
121	converted.add(convertRow(r));	1✔
122	}	1✔
123	}
124	return converted;	1✔
125	}
126
127	private void fillDataFromPrefetch() {
128	try {
129	if (currentIdxInBuffer >= currentBuffer.size() && 0 != currentIdxInBuffer) {	1✔
130	if (prefetchEnabled) {	1✔
131	if (!isBackupFilled) {	1✔
132	fetchFuture.get();	×
133	}
134	// copy backup buffer to current buffer and fill backup buffer asyn
135	currentIdxInBuffer = 0;	1✔
136	currentBuffer.clear();	1✔
137	currentBuffer = backupBuffer;	1✔
138	isBackupFilled = false;	1✔
139	fetchFuture = executorService.submit(new RowsFetcher(true));	1✔
140	} else {
141	currentIdxInBuffer = 0;	1✔
142	new RowsFetcher(false).call();	1✔
143	}
144	}
145	} catch (Exception e) {	×
146	throw new RuntimeException(e);	×
147	}	1✔
148	}	1✔
149
150	/**
151	* populate a row with index counter increased
152	*/
153	private void popRow() {
154	fillDataFromPrefetch();	1✔
155	currentRawRow = currentBuffer.get(currentIdxInBuffer);	1✔
156	currentIdxInBuffer++;	1✔
157	}	1✔
158
159	/**
160	* populate a row with index counter unchanged
161	*/
162	private void pickRow() {
163	fillDataFromPrefetch();	1✔
164	currentRawRow = currentBuffer.get(currentIdxInBuffer);	1✔
165	}	1✔
166
167	@Override
168	public boolean hasNext() {
169	fillDataFromPrefetch();	1✔
170	if (currentIdxInBuffer < currentBuffer.size()) {	1✔
171	return true;	1✔
172	}
173
174	return false;	1✔
175	}
176
177	@Override
178	public Object[] next() {
179	try {
180	popRow();	1✔
181	return this.currentRawRow;	1✔
182	} catch (Exception e) {	×
183	throw new RuntimeException(e);	×
184	}
185	}
186
187	/**
188	* for fetching the row with out incrementing counter.
189	* @return
190	*/
191	public Object[] fetchConverted() throws KeyGenException {
192	pickRow();	1✔
193	return this.currentRawRow;	1✔
194	}
195
196	private Object[] convertRow(Object[] rawRow) throws KeyGenException {
197	byte[] dims = ((ByteArrayWrapper) rawRow[0]).getDictionaryKey();	1✔
198	long[] keyArray = sourceSegProperties.getDimensionKeyGenerator().getKeyArray(dims);	1✔
199	byte[] covertedBytes =	1✔
200	destinationSegProperties.getDimensionKeyGenerator().generateKey(keyArray);	1✔
201	((ByteArrayWrapper) rawRow[0]).setDictionaryKey(covertedBytes);	1✔
202	return rawRow;	1✔
203	}
204
205	public void close() {
206	if (null != executorService) {	1✔
207	executorService.shutdownNow();	1✔
208	}
209	}	1✔
210	}

apache / carbondata / 2894

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