10501

Build # 10501

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/5ae16267efce6e454b670cf1b67319fb0cfa7126">5ae16267e<a href="https://github.com/apache/carbondata/commit/5ae16267efce6e454b670cf1b67319fb0cfa7126">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/783845ce124e9324ec6be806a82c0c6e48dd0807">783845ce1</a>">783845ce1</a><a href="https://github.com/apache/carbondata/commit/5ae16267efce6e454b670cf1b67319fb0cfa7126"> into 8e6def9fa">8e6def9fa</a>

Run Details

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

62503 of 78701 relevant lines covered (79.42%)

1.05 hits per line

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

76.64

/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SingleThreadFinalSortFilesMerger.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.sort.sortdata;

import java.io.File;
import java.io.FileFilter;
import java.util.AbstractQueue;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.PriorityQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

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.datastore.exception.CarbonDataWriterException;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.processing.loading.row.IntermediateSortTempRow;
import org.apache.carbondata.processing.loading.sort.SortStepRowHandler;
import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;

import org.apache.log4j.Logger;

public class SingleThreadFinalSortFilesMerger extends CarbonIterator<Object[]> {
  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(SingleThreadFinalSortFilesMerger.class.getName());

  /**
   * lockObject
   */
  private static final Object LOCKOBJECT = new Object();

  /**
   * recordHolderHeap
   */
  private AbstractQueue<SortTempFileChunkHolder> recordHolderHeapLocal;

  /**
   * tableName
   */
  private String tableName;
  private SortParameters sortParameters;
  private SortStepRowHandler sortStepRowHandler;
  /**
   * tempFileLocation
   */
  private String[] tempFileLocation;

  private int maxThreadForSorting;

  private ExecutorService executorService;

  private List<Future<Void>> mergerTask;

  public SingleThreadFinalSortFilesMerger(String[] tempFileLocation, String tableName,
      SortParameters sortParameters) {
    this.tempFileLocation = tempFileLocation;
    this.tableName = tableName;
    this.sortParameters = sortParameters;
    this.sortStepRowHandler = new SortStepRowHandler(sortParameters);
    try {
      maxThreadForSorting = Integer.parseInt(CarbonProperties.getInstance()
          .getProperty(CarbonCommonConstants.CARBON_MERGE_SORT_READER_THREAD,
              CarbonCommonConstants.CARBON_MERGE_SORT_READER_THREAD_DEFAULTVALUE));
    } catch (NumberFormatException e) {
      maxThreadForSorting =
          Integer.parseInt(CarbonCommonConstants.CARBON_MERGE_SORT_READER_THREAD_DEFAULTVALUE);
    }
    this.mergerTask = new ArrayList<>();
  }

  /**
   * This method will be used to merger the merged files
   *
   * @throws CarbonSortKeyAndGroupByException
   */
  public void startFinalMerge() throws CarbonDataWriterException {
    List<File> filesToMerge = getFilesToMergeSort();
    if (filesToMerge.size() == 0) {
      LOGGER.info("No file to merge in final merge stage");
      return;
    }

    startSorting(filesToMerge);
  }

  /**
   * Below method will be used to add in memory raw result iterator to priority queue.
   * This will be called in case of compaction, when it is compacting sorted and unsorted
   * both type of carbon data file
   * This method will add sorted file's RawResultIterator to priority queue using
   * InMemorySortTempChunkHolder as wrapper
   *
   * @param sortedRawResultMergerList
   * @param segmentProperties
   * @param noDicAndComplexColumns
   * @throws CarbonSortKeyAndGroupByException
   */
  public void addInMemoryRawResultIterator(List<RawResultIterator> sortedRawResultMergerList,
      SegmentProperties segmentProperties, CarbonColumn[] noDicAndComplexColumns,
      DataType[] measureDataType)
      throws CarbonSortKeyAndGroupByException {
    for (RawResultIterator rawResultIterator : sortedRawResultMergerList) {
      InMemorySortTempChunkHolder inMemorySortTempChunkHolder =
          new InMemorySortTempChunkHolder(rawResultIterator, segmentProperties,
              noDicAndComplexColumns, sortParameters, measureDataType);
      if (inMemorySortTempChunkHolder.hasNext()) {
        inMemorySortTempChunkHolder.readRow();
      }
      recordHolderHeapLocal.add(inMemorySortTempChunkHolder);
    }
  }

  private List<File> getFilesToMergeSort() {
    final int rangeId = sortParameters.getRangeId();
    FileFilter fileFilter = new FileFilter() {
      public boolean accept(File pathname) {
        return pathname.getName().startsWith(tableName + '_' + rangeId);
      }
    };

    // get all the merged files
    List<File> files = new ArrayList<File>(tempFileLocation.length);
    for (String tempLoc : tempFileLocation) {
      File[] subFiles = new File(tempLoc).listFiles(fileFilter);
      if (null != subFiles && subFiles.length > 0) {
        files.addAll(Arrays.asList(subFiles));
      }
    }

    return files;
  }

  /**
   * Below method will be used to start storing process This method will get
   * all the temp files present in sort temp folder then it will create the
   * record holder heap and then it will read first record from each file and
   * initialize the heap
   *
   * @throws CarbonSortKeyAndGroupByException
   */
  private void startSorting(List<File> files) throws CarbonDataWriterException {
    if (files.size() == 0) {
      LOGGER.info("No files to merge sort");
      return;
    }

    LOGGER.info("Started Final Merge");

    LOGGER.info("Number of temp file: " + files.size());

    // create record holder heap
    createRecordHolderQueue(files.size());

    // iterate over file list and create chunk holder and add to heap
    LOGGER.info("Started adding first record from each file");
    this.executorService = Executors.newFixedThreadPool(maxThreadForSorting);

    for (final File tempFile : files) {

      Callable<Void> callable = new Callable<Void>() {
        @Override public Void call() throws CarbonSortKeyAndGroupByException {
            // create chunk holder
            SortTempFileChunkHolder sortTempFileChunkHolder =
                new SortTempFileChunkHolder(tempFile, sortParameters, tableName, true);
          try {
            // initialize
            sortTempFileChunkHolder.initialize();
            sortTempFileChunkHolder.readRow();
          } catch (CarbonSortKeyAndGroupByException ex) {
            sortTempFileChunkHolder.closeStream();
            notifyFailure(ex);
          }
          synchronized (LOCKOBJECT) {
            recordHolderHeapLocal.add(sortTempFileChunkHolder);
          }
          return null;
        }
      };
      mergerTask.add(executorService.submit(callable));
    }
    executorService.shutdown();
    try {
      executorService.awaitTermination(2, TimeUnit.HOURS);
    } catch (Exception e) {
      throw new CarbonDataWriterException(e.getMessage(), e);
    }
    checkFailure();
    LOGGER.info("final merger Heap Size" + this.recordHolderHeapLocal.size());
  }

  private void checkFailure() {
    for (int i = 0; i < mergerTask.size(); i++) {
      try {
        mergerTask.get(i).get();
      } catch (InterruptedException | ExecutionException e) {
        throw new CarbonDataWriterException(e);
      }
    }
  }

  /**
   * This method will be used to create the heap which will be used to hold
   * the chunk of data
   */
  private void createRecordHolderQueue(int size) {
    // creating record holder heap
    this.recordHolderHeapLocal = new PriorityQueue<SortTempFileChunkHolder>(size);
  }

  private synchronized void notifyFailure(Throwable throwable) {
    close();
    LOGGER.error(throwable);
  }

  /**
   * This method will be used to get the sorted sort temp row from the sort temp files
   *
   * @return sorted row
   * @throws CarbonSortKeyAndGroupByException
   */
  public Object[] next() {
    if (hasNext()) {
      IntermediateSortTempRow sortTempRow = getSortedRecordFromFile();
      return sortStepRowHandler.convertIntermediateSortTempRowTo3Parted(sortTempRow);
    } else {
      throw new NoSuchElementException("No more elements to return");
    }
  }

  /**
   * This method will be used to get the sorted record from file
   *
   * @return sorted record sorted record
   * @throws CarbonSortKeyAndGroupByException
   */
  private IntermediateSortTempRow getSortedRecordFromFile() throws CarbonDataWriterException {
    IntermediateSortTempRow row = null;

    // poll the top object from heap
    // heap maintains binary tree which is based on heap condition that will
    // be based on comparator we are passing the heap
    // when will call poll it will always delete root of the tree and then
    // it does trickel down operation complexity is log(n)
    SortTempFileChunkHolder poll = this.recordHolderHeapLocal.poll();

    // get the row from chunk
    row = poll.getRow();

    // check if there no entry present
    if (!poll.hasNext()) {
      // if chunk is empty then close the stream
      poll.closeStream();

      // reaturn row
      return row;
    }

    // read new row
    try {
      poll.readRow();
    } catch (CarbonSortKeyAndGroupByException e) {
      close();
      throw new CarbonDataWriterException(e.getMessage(), e);
    }

    // add to heap
    this.recordHolderHeapLocal.add(poll);

    // return row
    return row;
  }

  /**
   * This method will be used to check whether any more element is present or
   * not
   *
   * @return more element is present
   */
  public boolean hasNext() {
    return this.recordHolderHeapLocal.size() > 0;
  }

  public void close() {
    if (null != executorService && !executorService.isShutdown()) {
      executorService.shutdownNow();
    }
    if (null != recordHolderHeapLocal) {
      SortTempFileChunkHolder sortTempFileChunkHolder;
      while (!recordHolderHeapLocal.isEmpty()) {
        sortTempFileChunkHolder = recordHolderHeapLocal.poll();
        if (null != sortTempFileChunkHolder) {
          sortTempFileChunkHolder.closeStream();
        }
      }
    }
  }
}

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
18	package org.apache.carbondata.processing.sort.sortdata;
19
20	import java.io.File;
21	import java.io.FileFilter;
22	import java.util.AbstractQueue;
23	import java.util.ArrayList;
24	import java.util.Arrays;
25	import java.util.List;
26	import java.util.NoSuchElementException;
27	import java.util.PriorityQueue;
28	import java.util.concurrent.Callable;
29	import java.util.concurrent.ExecutionException;
30	import java.util.concurrent.ExecutorService;
31	import java.util.concurrent.Executors;
32	import java.util.concurrent.Future;
33	import java.util.concurrent.TimeUnit;
34
35	import org.apache.carbondata.common.CarbonIterator;
36	import org.apache.carbondata.common.logging.LogServiceFactory;
37	import org.apache.carbondata.core.constants.CarbonCommonConstants;
38	import org.apache.carbondata.core.datastore.block.SegmentProperties;
39	import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
40	import org.apache.carbondata.core.metadata.datatype.DataType;
41	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
42	import org.apache.carbondata.core.scan.result.iterator.RawResultIterator;
43	import org.apache.carbondata.core.util.CarbonProperties;
44	import org.apache.carbondata.processing.loading.row.IntermediateSortTempRow;
45	import org.apache.carbondata.processing.loading.sort.SortStepRowHandler;
46	import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
47
48	import org.apache.log4j.Logger;
49
50	public class SingleThreadFinalSortFilesMerger extends CarbonIterator<Object[]> {
51	/**
52	* LOGGER
53	*/
54	private static final Logger LOGGER =	1✔
55	LogServiceFactory.getLogService(SingleThreadFinalSortFilesMerger.class.getName());	1✔
56
57	/**
58	* lockObject
59	*/
60	private static final Object LOCKOBJECT = new Object();	1✔
61
62	/**
63	* recordHolderHeap
64	*/
65	private AbstractQueue<SortTempFileChunkHolder> recordHolderHeapLocal;
66
67	/**
68	* tableName
69	*/
70	private String tableName;
71	private SortParameters sortParameters;
72	private SortStepRowHandler sortStepRowHandler;
73	/**
74	* tempFileLocation
75	*/
76	private String[] tempFileLocation;
77
78	private int maxThreadForSorting;
79
80	private ExecutorService executorService;
81
82	private List<Future<Void>> mergerTask;
83
84	public SingleThreadFinalSortFilesMerger(String[] tempFileLocation, String tableName,
85	SortParameters sortParameters) {	1✔
86	this.tempFileLocation = tempFileLocation;	1✔
87	this.tableName = tableName;	1✔
88	this.sortParameters = sortParameters;	1✔
89	this.sortStepRowHandler = new SortStepRowHandler(sortParameters);	1✔
90	try {
91	maxThreadForSorting = Integer.parseInt(CarbonProperties.getInstance()	1✔
92	.getProperty(CarbonCommonConstants.CARBON_MERGE_SORT_READER_THREAD,	1✔
93	CarbonCommonConstants.CARBON_MERGE_SORT_READER_THREAD_DEFAULTVALUE));
94	} catch (NumberFormatException e) {	×
95	maxThreadForSorting =	×
96	Integer.parseInt(CarbonCommonConstants.CARBON_MERGE_SORT_READER_THREAD_DEFAULTVALUE);	×
97	}	1✔
98	this.mergerTask = new ArrayList<>();	1✔
99	}	1✔
100
101	/**
102	* This method will be used to merger the merged files
103	*
104	* @throws CarbonSortKeyAndGroupByException
105	*/
106	public void startFinalMerge() throws CarbonDataWriterException {
107	List<File> filesToMerge = getFilesToMergeSort();	1✔
108	if (filesToMerge.size() == 0) {	1✔
109	LOGGER.info("No file to merge in final merge stage");	×
110	return;	×
111	}
112
113	startSorting(filesToMerge);	1✔
114	}	1✔
115
116	/**
117	* Below method will be used to add in memory raw result iterator to priority queue.
118	* This will be called in case of compaction, when it is compacting sorted and unsorted
119	* both type of carbon data file
120	* This method will add sorted file's RawResultIterator to priority queue using
121	* InMemorySortTempChunkHolder as wrapper
122	*
123	* @param sortedRawResultMergerList
124	* @param segmentProperties
125	* @param noDicAndComplexColumns
126	* @throws CarbonSortKeyAndGroupByException
127	*/
128	public void addInMemoryRawResultIterator(List<RawResultIterator> sortedRawResultMergerList,
129	SegmentProperties segmentProperties, CarbonColumn[] noDicAndComplexColumns,
130	DataType[] measureDataType)
131	throws CarbonSortKeyAndGroupByException {
132	for (RawResultIterator rawResultIterator : sortedRawResultMergerList) {	1✔
133	InMemorySortTempChunkHolder inMemorySortTempChunkHolder =	1✔
134	new InMemorySortTempChunkHolder(rawResultIterator, segmentProperties,
135	noDicAndComplexColumns, sortParameters, measureDataType);
136	if (inMemorySortTempChunkHolder.hasNext()) {	1✔
137	inMemorySortTempChunkHolder.readRow();	1✔
138	}
139	recordHolderHeapLocal.add(inMemorySortTempChunkHolder);	1✔
140	}	1✔
141	}	1✔
142
143	private List<File> getFilesToMergeSort() {
144	final int rangeId = sortParameters.getRangeId();	1✔
145	FileFilter fileFilter = new FileFilter() {	1✔
146	public boolean accept(File pathname) {
147	return pathname.getName().startsWith(tableName + '_' + rangeId);	1✔
148	}
149	};
150
151	// get all the merged files
152	List<File> files = new ArrayList<File>(tempFileLocation.length);	1✔
153	for (String tempLoc : tempFileLocation) {	1✔
154	File[] subFiles = new File(tempLoc).listFiles(fileFilter);	1✔
155	if (null != subFiles && subFiles.length > 0) {	1✔
156	files.addAll(Arrays.asList(subFiles));	1✔
157	}
158	}
159
160	return files;	1✔
161	}
162
163	/**
164	* Below method will be used to start storing process This method will get
165	* all the temp files present in sort temp folder then it will create the
166	* record holder heap and then it will read first record from each file and
167	* initialize the heap
168	*
169	* @throws CarbonSortKeyAndGroupByException
170	*/
171	private void startSorting(List<File> files) throws CarbonDataWriterException {
172	if (files.size() == 0) {	1✔
173	LOGGER.info("No files to merge sort");	×
174	return;	×
175	}
176
177	LOGGER.info("Started Final Merge");	1✔
178
179	LOGGER.info("Number of temp file: " + files.size());	1✔
180
181	// create record holder heap
182	createRecordHolderQueue(files.size());	1✔
183
184	// iterate over file list and create chunk holder and add to heap
185	LOGGER.info("Started adding first record from each file");	1✔
186	this.executorService = Executors.newFixedThreadPool(maxThreadForSorting);	1✔
187
188	for (final File tempFile : files) {	1✔
189
190	Callable<Void> callable = new Callable<Void>() {	1✔
191	@Override public Void call() throws CarbonSortKeyAndGroupByException {
192	// create chunk holder
193	SortTempFileChunkHolder sortTempFileChunkHolder =	1✔
194	new SortTempFileChunkHolder(tempFile, sortParameters, tableName, true);	1✔
195	try {
196	// initialize
197	sortTempFileChunkHolder.initialize();	1✔
198	sortTempFileChunkHolder.readRow();	1✔
199	} catch (CarbonSortKeyAndGroupByException ex) {	×
200	sortTempFileChunkHolder.closeStream();	×
201	notifyFailure(ex);	×
202	}	1✔
203	synchronized (LOCKOBJECT) {	1✔
204	recordHolderHeapLocal.add(sortTempFileChunkHolder);	1✔
205	}	1✔
206	return null;	1✔
207	}
208	};
209	mergerTask.add(executorService.submit(callable));	1✔
210	}	1✔
211	executorService.shutdown();	1✔
212	try {
213	executorService.awaitTermination(2, TimeUnit.HOURS);	1✔
214	} catch (Exception e) {	×
215	throw new CarbonDataWriterException(e.getMessage(), e);	×
216	}	1✔
217	checkFailure();	1✔
218	LOGGER.info("final merger Heap Size" + this.recordHolderHeapLocal.size());	1✔
219	}	1✔
220
221	private void checkFailure() {
222	for (int i = 0; i < mergerTask.size(); i++) {	1✔
223	try {
224	mergerTask.get(i).get();	1✔
225	} catch (InterruptedException \| ExecutionException e) {	×
226	throw new CarbonDataWriterException(e);	×
227	}	1✔
228	}
229	}	1✔
230
231	/**
232	* This method will be used to create the heap which will be used to hold
233	* the chunk of data
234	*/
235	private void createRecordHolderQueue(int size) {
236	// creating record holder heap
237	this.recordHolderHeapLocal = new PriorityQueue<SortTempFileChunkHolder>(size);	1✔
238	}	1✔
239
240	private synchronized void notifyFailure(Throwable throwable) {
241	close();	×
242	LOGGER.error(throwable);	×
243	}	×
244
245	/**
246	* This method will be used to get the sorted sort temp row from the sort temp files
247	*
248	* @return sorted row
249	* @throws CarbonSortKeyAndGroupByException
250	*/
251	public Object[] next() {
252	if (hasNext()) {	1✔
253	IntermediateSortTempRow sortTempRow = getSortedRecordFromFile();	1✔
254	return sortStepRowHandler.convertIntermediateSortTempRowTo3Parted(sortTempRow);	1✔
255	} else {
256	throw new NoSuchElementException("No more elements to return");	×
257	}
258	}
259
260	/**
261	* This method will be used to get the sorted record from file
262	*
263	* @return sorted record sorted record
264	* @throws CarbonSortKeyAndGroupByException
265	*/
266	private IntermediateSortTempRow getSortedRecordFromFile() throws CarbonDataWriterException {
267	IntermediateSortTempRow row = null;	1✔
268
269	// poll the top object from heap
270	// heap maintains binary tree which is based on heap condition that will
271	// be based on comparator we are passing the heap
272	// when will call poll it will always delete root of the tree and then
273	// it does trickel down operation complexity is log(n)
274	SortTempFileChunkHolder poll = this.recordHolderHeapLocal.poll();	1✔
275
276	// get the row from chunk
277	row = poll.getRow();	1✔
278
279	// check if there no entry present
280	if (!poll.hasNext()) {	1✔
281	// if chunk is empty then close the stream
282	poll.closeStream();	1✔
283
284	// reaturn row
285	return row;	1✔
286	}
287
288	// read new row
289	try {
290	poll.readRow();	1✔
291	} catch (CarbonSortKeyAndGroupByException e) {	×
292	close();	×
293	throw new CarbonDataWriterException(e.getMessage(), e);	×
294	}	1✔
295
296	// add to heap
297	this.recordHolderHeapLocal.add(poll);	1✔
298
299	// return row
300	return row;	1✔
301	}
302
303	/**
304	* This method will be used to check whether any more element is present or
305	* not
306	*
307	* @return more element is present
308	*/
309	public boolean hasNext() {
310	return this.recordHolderHeapLocal.size() > 0;	1✔
311	}
312
313	public void close() {
314	if (null != executorService && !executorService.isShutdown()) {	1✔
315	executorService.shutdownNow();	×
316	}
317	if (null != recordHolderHeapLocal) {	1✔
318	SortTempFileChunkHolder sortTempFileChunkHolder;
319	while (!recordHolderHeapLocal.isEmpty()) {	1✔
320	sortTempFileChunkHolder = recordHolderHeapLocal.poll();	×
321	if (null != sortTempFileChunkHolder) {	×
322	sortTempFileChunkHolder.closeStream();	×
323	}
324	}
325	}
326	}	1✔
327	}

apache / carbondata / 10501

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