2042

Build # 2042

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/dbd6590a49e514e014f49429e4f61edddf7e6e93">dbd6590a4<a href="https://github.com/apache/carbondata/commit/dbd6590a49e514e014f49429e4f61edddf7e6e93">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6b519122947a25040b2103b0a0283d8567fb1613">6b5191229</a>">6b5191229</a><a href="https://github.com/apache/carbondata/commit/dbd6590a49e514e014f49429e4f61edddf7e6e93"> into cabafe56d">cabafe56d</a>

Run Details

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

61379 of 77778 relevant lines covered (78.92%)

1.05 hits per line

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

87.8

/processing/src/main/java/org/apache/carbondata/processing/loading/sort/impl/ParallelReadMergeSorterWithColumnRangeImpl.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.processing.loading.sort.impl;

import java.io.File;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

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.exception.CarbonDataWriterException;
import org.apache.carbondata.core.datastore.row.CarbonRow;
import org.apache.carbondata.core.metadata.schema.ColumnRangeInfo;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory;
import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
import org.apache.carbondata.processing.loading.row.CarbonRowBatch;
import org.apache.carbondata.processing.loading.sort.AbstractMergeSorter;
import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
import org.apache.carbondata.processing.sort.sortdata.SingleThreadFinalSortFilesMerger;
import org.apache.carbondata.processing.sort.sortdata.SortDataRows;
import org.apache.carbondata.processing.sort.sortdata.SortIntermediateFileMerger;
import org.apache.carbondata.processing.sort.sortdata.SortParameters;
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;

import org.apache.log4j.Logger;

/**
 * It parallely reads data from array of iterates and do merge sort.
 * First it sorts the data and write to temp files. These temp files will be merge sorted to get
 * final merge sort result.
 * This step is specifically for the data loading with specifying column value range, such as
 * bucketing,sort_column_bounds, it sorts each range of data separately and write to temp files.
 */
public class ParallelReadMergeSorterWithColumnRangeImpl extends AbstractMergeSorter {
  private static final Logger LOGGER = LogServiceFactory.getLogService(
      ParallelReadMergeSorterWithColumnRangeImpl.class.getName());

  private SortParameters originSortParameters;

  private SortIntermediateFileMerger[] intermediateFileMergers;

  private ColumnRangeInfo columnRangeInfo;

  private int sortBufferSize;

  private AtomicLong rowCounter;
  /**
   * counters to collect information about rows processed by each range
   */
  private List<AtomicLong> insideRowCounterList;

  public ParallelReadMergeSorterWithColumnRangeImpl(AtomicLong rowCounter,
      ColumnRangeInfo columnRangeInfo) {
    this.rowCounter = rowCounter;
    this.columnRangeInfo = columnRangeInfo;
  }

  @Override
  public void initialize(SortParameters sortParameters) {
    this.originSortParameters = sortParameters;
    int buffer = Integer.parseInt(CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.SORT_SIZE, CarbonCommonConstants.SORT_SIZE_DEFAULT_VAL));
    sortBufferSize = buffer / columnRangeInfo.getNumOfRanges();
    if (sortBufferSize < 100) {
      sortBufferSize = 100;
    }
    this.insideRowCounterList = new ArrayList<>(columnRangeInfo.getNumOfRanges());
    for (int i = 0; i < columnRangeInfo.getNumOfRanges(); i++) {
      insideRowCounterList.add(new AtomicLong(0));
    }
  }

  @Override public Iterator<CarbonRowBatch>[] sort(Iterator<CarbonRowBatch>[] iterators)
      throws CarbonDataLoadingException {
    SortDataRows[] sortDataRows = new SortDataRows[columnRangeInfo.getNumOfRanges()];
    intermediateFileMergers = new SortIntermediateFileMerger[columnRangeInfo.getNumOfRanges()];
    SortParameters[] sortParameterArray = new SortParameters[columnRangeInfo.getNumOfRanges()];
    try {
      for (int i = 0; i < columnRangeInfo.getNumOfRanges(); i++) {
        SortParameters parameters = originSortParameters.getCopy();
        parameters.setPartitionID(i + "");
        parameters.setRangeId(i);
        sortParameterArray[i] = parameters;
        setTempLocation(parameters);
        parameters.setBufferSize(sortBufferSize);
        intermediateFileMergers[i] = new SortIntermediateFileMerger(parameters);
        sortDataRows[i] = new SortDataRows(parameters, intermediateFileMergers[i]);
        sortDataRows[i].initialize();
      }
    } catch (CarbonSortKeyAndGroupByException e) {
      throw new CarbonDataLoadingException(e);
    }
    ExecutorService executorService = Executors.newFixedThreadPool(iterators.length);
    this.threadStatusObserver = new ThreadStatusObserver(executorService);
    final int batchSize = CarbonProperties.getInstance().getBatchSize();
    try {
      // dispatch rows to sortDataRows by range id
      for (int i = 0; i < iterators.length; i++) {
        executorService.execute(new SortIteratorThread(iterators[i], sortDataRows, rowCounter,
            this.insideRowCounterList, this.threadStatusObserver));
      }
      executorService.shutdown();
      executorService.awaitTermination(2, TimeUnit.DAYS);
      processRowToNextStep(sortDataRows, originSortParameters);
    } catch (Exception e) {
      checkError();
      throw new CarbonDataLoadingException("Problem while shutdown the server ", e);
    }
    checkError();
    try {
      for (int i = 0; i < intermediateFileMergers.length; i++) {
        intermediateFileMergers[i].finish();
      }
    } catch (CarbonDataWriterException e) {
      throw new CarbonDataLoadingException(e);
    } catch (CarbonSortKeyAndGroupByException e) {
      throw new CarbonDataLoadingException(e);
    }

    Iterator<CarbonRowBatch>[] batchIterator = new Iterator[columnRangeInfo.getNumOfRanges()];
    for (int i = 0; i < columnRangeInfo.getNumOfRanges(); i++) {
      batchIterator[i] = new MergedDataIterator(sortParameterArray[i], batchSize);
    }

    return batchIterator;
  }

  private SingleThreadFinalSortFilesMerger getFinalMerger(SortParameters sortParameters) {
    String[] storeLocation = CarbonDataProcessorUtil
        .getLocalDataFolderLocation(sortParameters.getCarbonTable(),
            String.valueOf(sortParameters.getTaskNo()),
            sortParameters.getSegmentId() + "", false, false);
    // Set the data file location
    String[] dataFolderLocation = CarbonDataProcessorUtil.arrayAppend(storeLocation, File.separator,
        CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
    return new SingleThreadFinalSortFilesMerger(dataFolderLocation, sortParameters.getTableName(),
        sortParameters);
  }

  @Override public void close() {
    for (int i = 0; i < intermediateFileMergers.length; i++) {
      intermediateFileMergers[i].close();
    }
  }

  /**
   * Below method will be used to process data to next step
   */
  private boolean processRowToNextStep(SortDataRows[] sortDataRows, SortParameters parameters)
      throws CarbonDataLoadingException {
    try {
      for (int i = 0; i < sortDataRows.length; i++) {
        // start sorting
        sortDataRows[i].startSorting();
      }
      // check any more rows are present
      LOGGER.info("Record Processed For table: " + parameters.getTableName());
      CarbonTimeStatisticsFactory.getLoadStatisticsInstance()
          .recordSortRowsStepTotalTime(parameters.getPartitionID(), System.currentTimeMillis());
      CarbonTimeStatisticsFactory.getLoadStatisticsInstance()
          .recordDictionaryValuesTotalTime(parameters.getPartitionID(), System.currentTimeMillis());
      return false;
    } catch (CarbonSortKeyAndGroupByException e) {
      throw new CarbonDataLoadingException(e);
    }
  }

  private void setTempLocation(SortParameters parameters) {
    String[] carbonDataDirectoryPath = CarbonDataProcessorUtil
        .getLocalDataFolderLocation(parameters.getCarbonTable(), parameters.getTaskNo(),
            parameters.getSegmentId(),
            false, false);
    String[] tmpLocs = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath, File.separator,
        CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
    parameters.setTempFileLocation(tmpLocs);
  }

  /**
   * This thread iterates the iterator and adds the rows to @{@link SortDataRows}
   */
  private static class SortIteratorThread implements Runnable {

    private Iterator<CarbonRowBatch> iterator;

    private SortDataRows[] sortDataRows;

    private AtomicLong rowCounter;
    private List<AtomicLong> insideCounterList;
    private ThreadStatusObserver threadStatusObserver;

    public SortIteratorThread(Iterator<CarbonRowBatch> iterator, SortDataRows[] sortDataRows,
        AtomicLong rowCounter, List<AtomicLong> insideCounterList,
        ThreadStatusObserver observer) {
      this.iterator = iterator;
      this.sortDataRows = sortDataRows;
      this.rowCounter = rowCounter;
      this.insideCounterList = insideCounterList;
      this.threadStatusObserver = observer;
    }

    @Override
    public void run() {
      try {
        while (iterator.hasNext()) {
          CarbonRowBatch batch = iterator.next();
          while (batch.hasNext()) {
            CarbonRow row = batch.next();
            if (row != null) {
              SortDataRows sortDataRow = sortDataRows[row.getRangeId()];
              synchronized (sortDataRow) {
                sortDataRow.addRow(row.getData());
                insideCounterList.get(row.getRangeId()).getAndIncrement();
                rowCounter.getAndAdd(1);
              }
            }
          }
        }
        LOGGER.info("Rows processed by each range: " + insideCounterList);
      } catch (Exception e) {
        LOGGER.error(e);
        this.threadStatusObserver.notifyFailed(e);
      }
    }

  }

  private class MergedDataIterator extends CarbonIterator<CarbonRowBatch> {

    private SortParameters sortParameters;

    private int batchSize;

    private boolean firstRow = true;

    public MergedDataIterator(SortParameters sortParameters, int batchSize) {
      this.sortParameters = sortParameters;
      this.batchSize = batchSize;
    }

    private SingleThreadFinalSortFilesMerger finalMerger;

    @Override public boolean hasNext() {
      if (firstRow) {
        firstRow = false;
        finalMerger = getFinalMerger(sortParameters);
        finalMerger.startFinalMerge();
      }
      return finalMerger.hasNext();
    }

    @Override public CarbonRowBatch next() {
      int counter = 0;
      CarbonRowBatch rowBatch = new CarbonRowBatch(batchSize);
      while (finalMerger.hasNext() && counter < batchSize) {
        rowBatch.addRow(new CarbonRow(finalMerger.next()));
        counter++;
      }
      return rowBatch;
    }
  }
}

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.processing.loading.sort.impl;
18
19	import java.io.File;
20	import java.util.ArrayList;
21	import java.util.Iterator;
22	import java.util.List;
23	import java.util.concurrent.ExecutorService;
24	import java.util.concurrent.Executors;
25	import java.util.concurrent.TimeUnit;
26	import java.util.concurrent.atomic.AtomicLong;
27
28	import org.apache.carbondata.common.CarbonIterator;
29	import org.apache.carbondata.common.logging.LogServiceFactory;
30	import org.apache.carbondata.core.constants.CarbonCommonConstants;
31	import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
32	import org.apache.carbondata.core.datastore.row.CarbonRow;
33	import org.apache.carbondata.core.metadata.schema.ColumnRangeInfo;
34	import org.apache.carbondata.core.util.CarbonProperties;
35	import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory;
36	import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
37	import org.apache.carbondata.processing.loading.row.CarbonRowBatch;
38	import org.apache.carbondata.processing.loading.sort.AbstractMergeSorter;
39	import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
40	import org.apache.carbondata.processing.sort.sortdata.SingleThreadFinalSortFilesMerger;
41	import org.apache.carbondata.processing.sort.sortdata.SortDataRows;
42	import org.apache.carbondata.processing.sort.sortdata.SortIntermediateFileMerger;
43	import org.apache.carbondata.processing.sort.sortdata.SortParameters;
44	import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
45
46	import org.apache.log4j.Logger;
47
48	/**
49	* It parallely reads data from array of iterates and do merge sort.
50	* First it sorts the data and write to temp files. These temp files will be merge sorted to get
51	* final merge sort result.
52	* This step is specifically for the data loading with specifying column value range, such as
53	* bucketing,sort_column_bounds, it sorts each range of data separately and write to temp files.
54	*/
55	public class ParallelReadMergeSorterWithColumnRangeImpl extends AbstractMergeSorter {
56	private static final Logger LOGGER = LogServiceFactory.getLogService(	1✔
57	ParallelReadMergeSorterWithColumnRangeImpl.class.getName());	1✔
58
59	private SortParameters originSortParameters;
60
61	private SortIntermediateFileMerger[] intermediateFileMergers;
62
63	private ColumnRangeInfo columnRangeInfo;
64
65	private int sortBufferSize;
66
67	private AtomicLong rowCounter;
68	/**
69	* counters to collect information about rows processed by each range
70	*/
71	private List<AtomicLong> insideRowCounterList;
72
73	public ParallelReadMergeSorterWithColumnRangeImpl(AtomicLong rowCounter,
74	ColumnRangeInfo columnRangeInfo) {	1✔
75	this.rowCounter = rowCounter;	1✔
76	this.columnRangeInfo = columnRangeInfo;	1✔
77	}	1✔
78
79	@Override
80	public void initialize(SortParameters sortParameters) {
81	this.originSortParameters = sortParameters;	1✔
82	int buffer = Integer.parseInt(CarbonProperties.getInstance()	1✔
83	.getProperty(CarbonCommonConstants.SORT_SIZE, CarbonCommonConstants.SORT_SIZE_DEFAULT_VAL));	1✔
84	sortBufferSize = buffer / columnRangeInfo.getNumOfRanges();	1✔
85	if (sortBufferSize < 100) {	1✔
86	sortBufferSize = 100;	×
87	}
88	this.insideRowCounterList = new ArrayList<>(columnRangeInfo.getNumOfRanges());	1✔
89	for (int i = 0; i < columnRangeInfo.getNumOfRanges(); i++) {	1✔
90	insideRowCounterList.add(new AtomicLong(0));	1✔
91	}
92	}	1✔
93
94	@Override public Iterator<CarbonRowBatch>[] sort(Iterator<CarbonRowBatch>[] iterators)
95	throws CarbonDataLoadingException {
96	SortDataRows[] sortDataRows = new SortDataRows[columnRangeInfo.getNumOfRanges()];	1✔
97	intermediateFileMergers = new SortIntermediateFileMerger[columnRangeInfo.getNumOfRanges()];	1✔
98	SortParameters[] sortParameterArray = new SortParameters[columnRangeInfo.getNumOfRanges()];	1✔
99	try {
100	for (int i = 0; i < columnRangeInfo.getNumOfRanges(); i++) {	1✔
101	SortParameters parameters = originSortParameters.getCopy();	1✔
102	parameters.setPartitionID(i + "");	1✔
103	parameters.setRangeId(i);	1✔
104	sortParameterArray[i] = parameters;	1✔
105	setTempLocation(parameters);	1✔
106	parameters.setBufferSize(sortBufferSize);	1✔
107	intermediateFileMergers[i] = new SortIntermediateFileMerger(parameters);	1✔
108	sortDataRows[i] = new SortDataRows(parameters, intermediateFileMergers[i]);	1✔
109	sortDataRows[i].initialize();	1✔
110	}
111	} catch (CarbonSortKeyAndGroupByException e) {	×
112	throw new CarbonDataLoadingException(e);	×
113	}	1✔
114	ExecutorService executorService = Executors.newFixedThreadPool(iterators.length);	1✔
115	this.threadStatusObserver = new ThreadStatusObserver(executorService);	1✔
116	final int batchSize = CarbonProperties.getInstance().getBatchSize();	1✔
117	try {
118	// dispatch rows to sortDataRows by range id
119	for (int i = 0; i < iterators.length; i++) {	1✔
120	executorService.execute(new SortIteratorThread(iterators[i], sortDataRows, rowCounter,	1✔
121	this.insideRowCounterList, this.threadStatusObserver));
122	}
123	executorService.shutdown();	1✔
124	executorService.awaitTermination(2, TimeUnit.DAYS);	1✔
125	processRowToNextStep(sortDataRows, originSortParameters);	1✔
126	} catch (Exception e) {	×
127	checkError();	×
128	throw new CarbonDataLoadingException("Problem while shutdown the server ", e);	×
129	}	1✔
130	checkError();	1✔
131	try {
132	for (int i = 0; i < intermediateFileMergers.length; i++) {	1✔
133	intermediateFileMergers[i].finish();	1✔
134	}
135	} catch (CarbonDataWriterException e) {	×
136	throw new CarbonDataLoadingException(e);	×
137	} catch (CarbonSortKeyAndGroupByException e) {	×
138	throw new CarbonDataLoadingException(e);	×
139	}	1✔
140
141	Iterator<CarbonRowBatch>[] batchIterator = new Iterator[columnRangeInfo.getNumOfRanges()];	1✔
142	for (int i = 0; i < columnRangeInfo.getNumOfRanges(); i++) {	1✔
143	batchIterator[i] = new MergedDataIterator(sortParameterArray[i], batchSize);	1✔
144	}
145
146	return batchIterator;	1✔
147	}
148
149	private SingleThreadFinalSortFilesMerger getFinalMerger(SortParameters sortParameters) {
150	String[] storeLocation = CarbonDataProcessorUtil	1✔
151	.getLocalDataFolderLocation(sortParameters.getCarbonTable(),	1✔
152	String.valueOf(sortParameters.getTaskNo()),	1✔
153	sortParameters.getSegmentId() + "", false, false);	1✔
154	// Set the data file location
155	String[] dataFolderLocation = CarbonDataProcessorUtil.arrayAppend(storeLocation, File.separator,	1✔
156	CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
157	return new SingleThreadFinalSortFilesMerger(dataFolderLocation, sortParameters.getTableName(),	1✔
158	sortParameters);
159	}
160
161	@Override public void close() {
162	for (int i = 0; i < intermediateFileMergers.length; i++) {	1✔
163	intermediateFileMergers[i].close();	1✔
164	}
165	}	1✔
166
167	/**
168	* Below method will be used to process data to next step
169	*/
170	private boolean processRowToNextStep(SortDataRows[] sortDataRows, SortParameters parameters)
171	throws CarbonDataLoadingException {
172	try {
173	for (int i = 0; i < sortDataRows.length; i++) {	1✔
174	// start sorting
175	sortDataRows[i].startSorting();	1✔
176	}
177	// check any more rows are present
178	LOGGER.info("Record Processed For table: " + parameters.getTableName());	1✔
179	CarbonTimeStatisticsFactory.getLoadStatisticsInstance()	1✔
180	.recordSortRowsStepTotalTime(parameters.getPartitionID(), System.currentTimeMillis());	1✔
181	CarbonTimeStatisticsFactory.getLoadStatisticsInstance()	1✔
182	.recordDictionaryValuesTotalTime(parameters.getPartitionID(), System.currentTimeMillis());	1✔
183	return false;	1✔
184	} catch (CarbonSortKeyAndGroupByException e) {	×
185	throw new CarbonDataLoadingException(e);	×
186	}
187	}
188
189	private void setTempLocation(SortParameters parameters) {
190	String[] carbonDataDirectoryPath = CarbonDataProcessorUtil	1✔
191	.getLocalDataFolderLocation(parameters.getCarbonTable(), parameters.getTaskNo(),	1✔
192	parameters.getSegmentId(),	1✔
193	false, false);
194	String[] tmpLocs = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath, File.separator,	1✔
195	CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
196	parameters.setTempFileLocation(tmpLocs);	1✔
197	}	1✔
198
199	/**
200	* This thread iterates the iterator and adds the rows to @{@link SortDataRows}
201	*/
202	private static class SortIteratorThread implements Runnable {
203
204	private Iterator<CarbonRowBatch> iterator;
205
206	private SortDataRows[] sortDataRows;
207
208	private AtomicLong rowCounter;
209	private List<AtomicLong> insideCounterList;
210	private ThreadStatusObserver threadStatusObserver;
211
212	public SortIteratorThread(Iterator<CarbonRowBatch> iterator, SortDataRows[] sortDataRows,
213	AtomicLong rowCounter, List<AtomicLong> insideCounterList,
214	ThreadStatusObserver observer) {	1✔
215	this.iterator = iterator;	1✔
216	this.sortDataRows = sortDataRows;	1✔
217	this.rowCounter = rowCounter;	1✔
218	this.insideCounterList = insideCounterList;	1✔
219	this.threadStatusObserver = observer;	1✔
220	}	1✔
221
222	@Override
223	public void run() {
224	try {
225	while (iterator.hasNext()) {	1✔
226	CarbonRowBatch batch = iterator.next();	1✔
227	while (batch.hasNext()) {	1✔
228	CarbonRow row = batch.next();	1✔
229	if (row != null) {	1✔
230	SortDataRows sortDataRow = sortDataRows[row.getRangeId()];	1✔
231	synchronized (sortDataRow) {	1✔
232	sortDataRow.addRow(row.getData());	1✔
233	insideCounterList.get(row.getRangeId()).getAndIncrement();	1✔
234	rowCounter.getAndAdd(1);	1✔
235	}	1✔
236	}
237	}	1✔
238	}	1✔
239	LOGGER.info("Rows processed by each range: " + insideCounterList);	1✔
240	} catch (Exception e) {	×
241	LOGGER.error(e);	×
242	this.threadStatusObserver.notifyFailed(e);	×
243	}	1✔
244	}	1✔
245
246	}
247
248	private class MergedDataIterator extends CarbonIterator<CarbonRowBatch> {
249
250	private SortParameters sortParameters;
251
252	private int batchSize;
253
254	private boolean firstRow = true;	1✔
255
256	public MergedDataIterator(SortParameters sortParameters, int batchSize) {	1✔
257	this.sortParameters = sortParameters;	1✔
258	this.batchSize = batchSize;	1✔
259	}	1✔
260
261	private SingleThreadFinalSortFilesMerger finalMerger;
262
263	@Override public boolean hasNext() {
264	if (firstRow) {	1✔
265	firstRow = false;	1✔
266	finalMerger = getFinalMerger(sortParameters);	1✔
267	finalMerger.startFinalMerge();	1✔
268	}
269	return finalMerger.hasNext();	1✔
270	}
271
272	@Override public CarbonRowBatch next() {
273	int counter = 0;	1✔
274	CarbonRowBatch rowBatch = new CarbonRowBatch(batchSize);	1✔
275	while (finalMerger.hasNext() && counter < batchSize) {	1✔
276	rowBatch.addRow(new CarbonRow(finalMerger.next()));	1✔
277	counter++;	1✔
278	}
279	return rowBatch;	1✔
280	}
281	}
282	}

apache / carbondata / 2042

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