2908

Build # 2908

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/48d9f395f9208ed7671028a5c6eca793c88de298">48d9f395f<a href="https://github.com/apache/carbondata/commit/48d9f395f9208ed7671028a5c6eca793c88de298">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/b25d8f24862732e0b4f136b1eed6cd16265e7c76">b25d8f248</a>">b25d8f248</a><a href="https://github.com/apache/carbondata/commit/48d9f395f9208ed7671028a5c6eca793c88de298"> into 6840a1836">6840a1836</a>

Run Details

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

62775 of 79279 relevant lines covered (79.18%)

1.05 hits per line

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

0.0

/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/IntermediateFileMerger.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.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.AbstractQueue;
import java.util.NoSuchElementException;
import java.util.PriorityQueue;
import java.util.concurrent.Callable;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.util.CarbonUtil;
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 IntermediateFileMerger implements Callable<Void> {
  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(IntermediateFileMerger.class.getName());

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

  /**
   * fileCounter
   */
  private int fileCounter;

  /**
   * stream
   */
  private DataOutputStream stream;

  /**
   * totalNumberOfRecords
   */
  private int totalNumberOfRecords;

  private SortParameters mergerParameters;

  private File[] intermediateFiles;

  private File outPutFile;
  private int writeBufferSize;
  private String compressorName;

  private Throwable throwable;
  private TableFieldStat tableFieldStat;
  private SortStepRowHandler sortStepRowHandler;
  /**
   * IntermediateFileMerger Constructor
   */
  public IntermediateFileMerger(SortParameters mergerParameters, File[] intermediateFiles,
      File outPutFile) {
    this.mergerParameters = mergerParameters;
    this.fileCounter = intermediateFiles.length;
    this.intermediateFiles = intermediateFiles;
    this.outPutFile = outPutFile;
    this.writeBufferSize = mergerParameters.getBufferSize();
    this.compressorName = mergerParameters.getSortTempCompressorName();
    this.tableFieldStat = new TableFieldStat(mergerParameters);
    this.sortStepRowHandler = new SortStepRowHandler(tableFieldStat);
  }

  @Override public Void call() throws Exception {
    long intermediateMergeStartTime = System.currentTimeMillis();
    int fileConterConst = fileCounter;
    try {
      startSorting();
      initialize();
      while (hasNext()) {
        writeDataToFile(next());
      }
      double intermediateMergeCostTime =
          (System.currentTimeMillis() - intermediateMergeStartTime) / 1000.0;
      LOGGER.info("============================== Intermediate Merge of " + fileConterConst +
          " Sort Temp Files Cost Time: " + intermediateMergeCostTime + "(s)");
    } catch (Exception e) {
      LOGGER.error("Problem while intermediate merging", e);
      clear();
      throwable = e;
    } finally {
      CarbonUtil.closeStreams(this.stream);
      if (null == throwable) {
        try {
          finish();
        } catch (CarbonSortKeyAndGroupByException e) {
          LOGGER.error("Problem while deleting the merge file", e);
          throwable = e;
        }
      } else {
        if (!outPutFile.delete()) {
          LOGGER.error("Problem while deleting the merge file");
        }
      }
    }
    if (null != throwable) {
      throw new CarbonSortKeyAndGroupByException(throwable);
    }
    return null;
  }

  /**
   * This method is responsible for initializing the out stream
   *
   * @throws CarbonSortKeyAndGroupByException
   */
  private void initialize() throws CarbonSortKeyAndGroupByException {
    try {
      stream = FileFactory.getDataOutputStream(outPutFile.getPath(), FileFactory.FileType.LOCAL,
          writeBufferSize, compressorName);
      this.stream.writeInt(this.totalNumberOfRecords);
    } catch (FileNotFoundException e) {
      throw new CarbonSortKeyAndGroupByException("Problem while getting the file", e);
    } catch (IOException e) {
      throw new CarbonSortKeyAndGroupByException("Problem while writing the data to file", e);
    }
  }

  /**
   * This method will be used to get the sorted sort temp row from sort temp file
   *
   * @return sorted record sorted record
   * @throws CarbonSortKeyAndGroupByException
   */
  private IntermediateSortTempRow getSortedRecordFromFile()
      throws CarbonSortKeyAndGroupByException {
    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.recordHolderHeap.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();

      // change the file counter
      --this.fileCounter;

      // reaturn row
      return row;
    }

    // read new row
    poll.readRow();

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

    // return row
    return row;
  }

  /**
   * 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() throws CarbonSortKeyAndGroupByException {
    LOGGER.info("Number of temp file: " + this.fileCounter);

    // create record holder heap
    createRecordHolderQueue(intermediateFiles);

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

    SortTempFileChunkHolder sortTempFileChunkHolder = null;

    for (File tempFile : intermediateFiles) {
      // create chunk holder
      sortTempFileChunkHolder =
          new SortTempFileChunkHolder(tempFile, mergerParameters, mergerParameters.getTableName(),
              false);

      // initialize
      sortTempFileChunkHolder.initialize();
      sortTempFileChunkHolder.readRow();
      this.totalNumberOfRecords += sortTempFileChunkHolder.getEntryCount();

      // add to heap
      this.recordHolderHeap.add(sortTempFileChunkHolder);
    }

    LOGGER.info("Heap Size: " + this.recordHolderHeap.size());
  }

  /**
   * This method will be used to create the heap which will be used to hold
   * the chunk of data
   *
   * @param listFiles list of temp files
   */
  private void createRecordHolderQueue(File[] listFiles) {
    // creating record holder heap
    this.recordHolderHeap = new PriorityQueue<>(listFiles.length);
  }

  /**
   * This method will be used to get the sorted sort temp row
   *
   * @return sorted row
   * @throws CarbonSortKeyAndGroupByException
   */
  private IntermediateSortTempRow next() throws CarbonSortKeyAndGroupByException {
    if (hasNext()) {
      return getSortedRecordFromFile();
    } else {
      throw new NoSuchElementException("No more elements to return");
    }
  }

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

  /**
   * Below method will be used to write data to file
   *
   * @throws IOException problem while writing
   */
  private void writeDataToFile(IntermediateSortTempRow row) throws IOException {
    sortStepRowHandler.writeIntermediateSortTempRowToOutputStream(row, stream);
  }

  private void finish() throws CarbonSortKeyAndGroupByException {
    clear();
    try {
      CarbonUtil.deleteFiles(intermediateFiles);
    } catch (IOException e) {
      throw new CarbonSortKeyAndGroupByException("Problem while deleting the intermediate files");
    }
  }

  private void clear() {
    if (recordHolderHeap != null) {
      SortTempFileChunkHolder sortTempFileChunkHolder;
      while (!recordHolderHeap.isEmpty()) {
        sortTempFileChunkHolder = recordHolderHeap.poll();
        if (null != sortTempFileChunkHolder) {
          sortTempFileChunkHolder.closeStream();
        }
      }
    }
    recordHolderHeap = null;
  }
}

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.DataOutputStream;
21	import java.io.File;
22	import java.io.FileNotFoundException;
23	import java.io.IOException;
24	import java.util.AbstractQueue;
25	import java.util.NoSuchElementException;
26	import java.util.PriorityQueue;
27	import java.util.concurrent.Callable;
28
29	import org.apache.carbondata.common.logging.LogServiceFactory;
30	import org.apache.carbondata.core.datastore.impl.FileFactory;
31	import org.apache.carbondata.core.util.CarbonUtil;
32	import org.apache.carbondata.processing.loading.row.IntermediateSortTempRow;
33	import org.apache.carbondata.processing.loading.sort.SortStepRowHandler;
34	import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
35
36	import org.apache.log4j.Logger;
37
38	public class IntermediateFileMerger implements Callable<Void> {
39	/**
40	* LOGGER
41	*/
42	private static final Logger LOGGER =	×
43	LogServiceFactory.getLogService(IntermediateFileMerger.class.getName());	×
44
45	/**
46	* recordHolderHeap
47	*/
48	private AbstractQueue<SortTempFileChunkHolder> recordHolderHeap;
49
50	/**
51	* fileCounter
52	*/
53	private int fileCounter;
54
55	/**
56	* stream
57	*/
58	private DataOutputStream stream;
59
60	/**
61	* totalNumberOfRecords
62	*/
63	private int totalNumberOfRecords;
64
65	private SortParameters mergerParameters;
66
67	private File[] intermediateFiles;
68
69	private File outPutFile;
70	private int writeBufferSize;
71	private String compressorName;
72
73	private Throwable throwable;
74	private TableFieldStat tableFieldStat;
75	private SortStepRowHandler sortStepRowHandler;
76	/**
77	* IntermediateFileMerger Constructor
78	*/
79	public IntermediateFileMerger(SortParameters mergerParameters, File[] intermediateFiles,
80	File outPutFile) {	×
81	this.mergerParameters = mergerParameters;	×
82	this.fileCounter = intermediateFiles.length;	×
83	this.intermediateFiles = intermediateFiles;	×
84	this.outPutFile = outPutFile;	×
85	this.writeBufferSize = mergerParameters.getBufferSize();	×
86	this.compressorName = mergerParameters.getSortTempCompressorName();	×
87	this.tableFieldStat = new TableFieldStat(mergerParameters);	×
88	this.sortStepRowHandler = new SortStepRowHandler(tableFieldStat);	×
89	}	×
90
91	@Override public Void call() throws Exception {
92	long intermediateMergeStartTime = System.currentTimeMillis();	×
93	int fileConterConst = fileCounter;	×
94	try {
95	startSorting();	×
96	initialize();	×
97	while (hasNext()) {	×
98	writeDataToFile(next());	×
99	}
100	double intermediateMergeCostTime =
101	(System.currentTimeMillis() - intermediateMergeStartTime) / 1000.0;	×
102	LOGGER.info("============================== Intermediate Merge of " + fileConterConst +	×
103	" Sort Temp Files Cost Time: " + intermediateMergeCostTime + "(s)");
104	} catch (Exception e) {	×
105	LOGGER.error("Problem while intermediate merging", e);	×
106	clear();	×
107	throwable = e;	×
108	} finally {
109	CarbonUtil.closeStreams(this.stream);	×
110	if (null == throwable) {	×
111	try {
112	finish();	×
113	} catch (CarbonSortKeyAndGroupByException e) {	×
114	LOGGER.error("Problem while deleting the merge file", e);	×
115	throwable = e;	×
116	}	×
117	} else {
118	if (!outPutFile.delete()) {	×
119	LOGGER.error("Problem while deleting the merge file");	×
120	}
121	}
122	}	×
123	if (null != throwable) {	×
124	throw new CarbonSortKeyAndGroupByException(throwable);	×
125	}
126	return null;	×
127	}
128
129	/**
130	* This method is responsible for initializing the out stream
131	*
132	* @throws CarbonSortKeyAndGroupByException
133	*/
134	private void initialize() throws CarbonSortKeyAndGroupByException {
135	try {
136	stream = FileFactory.getDataOutputStream(outPutFile.getPath(), FileFactory.FileType.LOCAL,	×
137	writeBufferSize, compressorName);
138	this.stream.writeInt(this.totalNumberOfRecords);	×
139	} catch (FileNotFoundException e) {	×
140	throw new CarbonSortKeyAndGroupByException("Problem while getting the file", e);	×
141	} catch (IOException e) {	×
142	throw new CarbonSortKeyAndGroupByException("Problem while writing the data to file", e);	×
143	}	×
144	}	×
145
146	/**
147	* This method will be used to get the sorted sort temp row from sort temp file
148	*
149	* @return sorted record sorted record
150	* @throws CarbonSortKeyAndGroupByException
151	*/
152	private IntermediateSortTempRow getSortedRecordFromFile()
153	throws CarbonSortKeyAndGroupByException {
154	IntermediateSortTempRow row = null;	×
155
156	// poll the top object from heap
157	// heap maintains binary tree which is based on heap condition that will
158	// be based on comparator we are passing the heap
159	// when will call poll it will always delete root of the tree and then
160	// it does trickel down operation complexity is log(n)
161	SortTempFileChunkHolder poll = this.recordHolderHeap.poll();	×
162
163	// get the row from chunk
164	row = poll.getRow();	×
165
166	// check if there no entry present
167	if (!poll.hasNext()) {	×
168	// if chunk is empty then close the stream
169	poll.closeStream();	×
170
171	// change the file counter
172	--this.fileCounter;	×
173
174	// reaturn row
175	return row;	×
176	}
177
178	// read new row
179	poll.readRow();	×
180
181	// add to heap
182	this.recordHolderHeap.add(poll);	×
183
184	// return row
185	return row;	×
186	}
187
188	/**
189	* Below method will be used to start storing process This method will get
190	* all the temp files present in sort temp folder then it will create the
191	* record holder heap and then it will read first record from each file and
192	* initialize the heap
193	*
194	* @throws CarbonSortKeyAndGroupByException
195	*/
196	private void startSorting() throws CarbonSortKeyAndGroupByException {
197	LOGGER.info("Number of temp file: " + this.fileCounter);	×
198
199	// create record holder heap
200	createRecordHolderQueue(intermediateFiles);	×
201
202	// iterate over file list and create chunk holder and add to heap
203	LOGGER.info("Started adding first record from each file");	×
204
205	SortTempFileChunkHolder sortTempFileChunkHolder = null;	×
206
207	for (File tempFile : intermediateFiles) {	×
208	// create chunk holder
209	sortTempFileChunkHolder =	×
210	new SortTempFileChunkHolder(tempFile, mergerParameters, mergerParameters.getTableName(),	×
211	false);
212
213	// initialize
214	sortTempFileChunkHolder.initialize();	×
215	sortTempFileChunkHolder.readRow();	×
216	this.totalNumberOfRecords += sortTempFileChunkHolder.getEntryCount();	×
217
218	// add to heap
219	this.recordHolderHeap.add(sortTempFileChunkHolder);	×
220	}
221
222	LOGGER.info("Heap Size: " + this.recordHolderHeap.size());	×
223	}	×
224
225	/**
226	* This method will be used to create the heap which will be used to hold
227	* the chunk of data
228	*
229	* @param listFiles list of temp files
230	*/
231	private void createRecordHolderQueue(File[] listFiles) {
232	// creating record holder heap
233	this.recordHolderHeap = new PriorityQueue<>(listFiles.length);	×
234	}	×
235
236	/**
237	* This method will be used to get the sorted sort temp row
238	*
239	* @return sorted row
240	* @throws CarbonSortKeyAndGroupByException
241	*/
242	private IntermediateSortTempRow next() throws CarbonSortKeyAndGroupByException {
243	if (hasNext()) {	×
244	return getSortedRecordFromFile();	×
245	} else {
246	throw new NoSuchElementException("No more elements to return");	×
247	}
248	}
249
250	/**
251	* This method will be used to check whether any more element is present or
252	* not
253	*
254	* @return more element is present
255	*/
256	private boolean hasNext() {
257	return this.fileCounter > 0;	×
258	}
259
260	/**
261	* Below method will be used to write data to file
262	*
263	* @throws IOException problem while writing
264	*/
265	private void writeDataToFile(IntermediateSortTempRow row) throws IOException {
266	sortStepRowHandler.writeIntermediateSortTempRowToOutputStream(row, stream);	×
267	}	×
268
269	private void finish() throws CarbonSortKeyAndGroupByException {
270	clear();	×
271	try {
272	CarbonUtil.deleteFiles(intermediateFiles);	×
273	} catch (IOException e) {	×
274	throw new CarbonSortKeyAndGroupByException("Problem while deleting the intermediate files");	×
275	}	×
276	}	×
277
278	private void clear() {
279	if (recordHolderHeap != null) {	×
280	SortTempFileChunkHolder sortTempFileChunkHolder;
281	while (!recordHolderHeap.isEmpty()) {	×
282	sortTempFileChunkHolder = recordHolderHeap.poll();	×
283	if (null != sortTempFileChunkHolder) {	×
284	sortTempFileChunkHolder.closeStream();	×
285	}
286	}
287	}
288	recordHolderHeap = null;	×
289	}	×
290	}

apache / carbondata / 2908

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