apache / carbondata / 291

Build # 291

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/63b2e1b202fecb7719815c05306c894a49df8e19">63b2e1b20<a href="https://github.com/apache/carbondata/commit/63b2e1b202fecb7719815c05306c894a49df8e19">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/5f5a96b5b76840993d11e72e2c07dfd23945d370">5f5a96b5b</a>">5f5a96b5b</a><a href="https://github.com/apache/carbondata/commit/63b2e1b202fecb7719815c05306c894a49df8e19"> into 61a5bd335">61a5bd335</a>

Run Details

7 of 7 new or added lines in 3 files covered. (100.0%)

65901 of 84032 relevant lines covered (78.42%)

2.07 hits per line

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

90.0

/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[]> {

  protected final SegmentProperties sourceSegProperties;

  protected 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;

  /**
   * number of cores which can be used
   */
  private int batchSize;
  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(RawResultIterator.class.getName());

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

  protected 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<>();
    while (detailRawQueryResultIterator.hasNext()) {
      for (Object[] r : detailRawQueryResultIterator.next().getRows()) {
        converted.add(convertRow(r));
      }
      if (converted.size() >= batchSize) {
        break;
      }
    }
    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;
  }

  protected 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	protected final SegmentProperties sourceSegProperties;
45
46	protected 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;	2✔
59	private boolean isBackupFilled = false;	2✔
60
61	/**
62	* number of cores which can be used
63	*/
64	private int batchSize;
65	/**
66	* LOGGER
67	*/
68	private static final Logger LOGGER =	2✔
69	LogServiceFactory.getLogService(RawResultIterator.class.getName());	2✔
70
71	public RawResultIterator(CarbonIterator<RowBatch> detailRawQueryResultIterator,
72	SegmentProperties sourceSegProperties, SegmentProperties destinationSegProperties,
73	boolean init) {	2✔
74	this.detailRawQueryResultIterator = detailRawQueryResultIterator;	2✔
75	this.sourceSegProperties = sourceSegProperties;	2✔
76	this.destinationSegProperties = destinationSegProperties;	2✔
77	this.executorService = Executors.newFixedThreadPool(1);	2✔
78	batchSize = CarbonProperties.getQueryBatchSize();	2✔
79	if (init) {	2✔
80	init();	2✔
81	}
82	}	2✔
83
84	protected void init() {
85	this.prefetchEnabled = CarbonProperties.getInstance().getProperty(	2✔
86	CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE,
87	CarbonCommonConstants.CARBON_COMPACTION_PREFETCH_ENABLE_DEFAULT).equalsIgnoreCase("true");	2✔
88	try {
89	new RowsFetcher(false).call();	2✔
90	if (prefetchEnabled) {	2✔
91	this.fetchFuture = executorService.submit(new RowsFetcher(true));	2✔
92	}
93	} catch (Exception e) {	×
94	LOGGER.error("Error occurs while fetching records", e);	×
95	throw new RuntimeException(e);	×
96	}	2✔
97	}	2✔
98
99	/**
100	* fetch rows
101	*/
102	private final class RowsFetcher implements Callable<Void> {
103	private boolean isBackupFilling;
104
105	private RowsFetcher(boolean isBackupFilling) {	2✔
106	this.isBackupFilling = isBackupFilling;	2✔
107	}	2✔
108
109	@Override
110	public Void call() throws Exception {
111	if (isBackupFilling) {	2✔
112	backupBuffer = fetchRows();	2✔
113	isBackupFilled = true;	2✔
114	} else {
115	currentBuffer = fetchRows();	2✔
116	}
117	return null;	2✔
118	}
119	}
120
121	private List<Object[]> fetchRows() throws Exception {
122	List<Object[]> converted = new ArrayList<>();	2✔
123	while (detailRawQueryResultIterator.hasNext()) {	2✔
124	for (Object[] r : detailRawQueryResultIterator.next().getRows()) {	2✔
125	converted.add(convertRow(r));	2✔
126	}	2✔
127	if (converted.size() >= batchSize) {	2✔
128	break;	2✔
129	}
130	}
131	return converted;	2✔
132	}
133
134	private void fillDataFromPrefetch() {
135	try {
136	if (currentIdxInBuffer >= currentBuffer.size() && 0 != currentIdxInBuffer) {	2✔
137	if (prefetchEnabled) {	2✔
138	if (!isBackupFilled) {	2✔
139	fetchFuture.get();	×
140	}
141	// copy backup buffer to current buffer and fill backup buffer asyn
142	currentIdxInBuffer = 0;	2✔
143	currentBuffer.clear();	2✔
144	currentBuffer = backupBuffer;	2✔
145	isBackupFilled = false;	2✔
146	fetchFuture = executorService.submit(new RowsFetcher(true));	2✔
147	} else {
148	currentIdxInBuffer = 0;	2✔
149	new RowsFetcher(false).call();	2✔
150	}
151	}
152	} catch (Exception e) {	×
153	throw new RuntimeException(e);	×
154	}	2✔
155	}	2✔
156
157	/**
158	* populate a row with index counter increased
159	*/
160	private void popRow() {
161	fillDataFromPrefetch();	2✔
162	currentRawRow = currentBuffer.get(currentIdxInBuffer);	2✔
163	currentIdxInBuffer++;	2✔
164	}	2✔
165
166	/**
167	* populate a row with index counter unchanged
168	*/
169	private void pickRow() {
170	fillDataFromPrefetch();	2✔
171	currentRawRow = currentBuffer.get(currentIdxInBuffer);	2✔
172	}	2✔
173
174	@Override
175	public boolean hasNext() {
176	fillDataFromPrefetch();	2✔
177	if (currentIdxInBuffer < currentBuffer.size()) {	2✔
178	return true;	2✔
179	}
180
181	return false;	2✔
182	}
183
184	@Override
185	public Object[] next() {
186	try {
187	popRow();	2✔
188	return this.currentRawRow;	2✔
189	} catch (Exception e) {	×
190	throw new RuntimeException(e);	×
191	}
192	}
193
194	/**
195	* for fetching the row with out incrementing counter.
196	* @return
197	*/
198	public Object[] fetchConverted() throws KeyGenException {
199	pickRow();	2✔
200	return this.currentRawRow;	2✔
201	}
202
203	protected Object[] convertRow(Object[] rawRow) throws KeyGenException {
204	byte[] dims = ((ByteArrayWrapper) rawRow[0]).getDictionaryKey();	2✔
205	long[] keyArray = sourceSegProperties.getDimensionKeyGenerator().getKeyArray(dims);	2✔
206	byte[] covertedBytes =	2✔
207	destinationSegProperties.getDimensionKeyGenerator().generateKey(keyArray);	2✔
208	((ByteArrayWrapper) rawRow[0]).setDictionaryKey(covertedBytes);	2✔
209	return rawRow;	2✔
210	}
211
212	public void close() {
213	if (null != executorService) {	2✔
214	executorService.shutdownNow();	2✔
215	}
216	}	2✔
217	}

apache / carbondata / 291

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