apache / carbondata / 913

Build # 913

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/616daa7053ae80dd4a0e87900cf414b65eb06713">616daa705">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/616daa7053ae80dd4a0e87900cf414b65eb06713">616daa705</a><a href="https://github.com/apache/carbondata/commit/616daa7053ae80dd4a0e87900cf414b65eb06713">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/616daa7053ae80dd4a0e87900cf414b65eb06713&quot;&gt;616daa705&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/616daa7053ae80dd4a0e87900cf414b65eb06713&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

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

60639 of 75226 relevant lines covered (80.61%)

1.06 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.LogService;
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;

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

apache / carbondata / 913

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