9102

Build # 9102

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/e47b7cf4d84ca751549df20e43e220afdf212a20">e47b7cf4d<a href="https://github.com/apache/carbondata/commit/e47b7cf4d84ca751549df20e43e220afdf212a20">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/e74d44d3648cc90335886cc5c525df2f5aa74ec4">e74d44d36</a>">e74d44d36</a><a href="https://github.com/apache/carbondata/commit/e47b7cf4d84ca751549df20e43e220afdf212a20"> into 15d38260c">15d38260c</a>

Run Details

484 of 484 new or added lines in 204 files covered. (100.0%)

60572 of 75108 relevant lines covered (80.65%)

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/loading/sort/unsafe/merger/UnsafeInMemoryIntermediateDataMerger.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.unsafe.merger;

import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.util.AbstractQueue;
import java.util.NoSuchElementException;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.concurrent.Callable;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
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.loading.sort.unsafe.UnsafeCarbonRowPage;
import org.apache.carbondata.processing.loading.sort.unsafe.holder.UnsafeCarbonRowForMerge;
import org.apache.carbondata.processing.loading.sort.unsafe.holder.UnsafeInmemoryMergeHolder;
import org.apache.carbondata.processing.sort.exception.CarbonSortKeyAndGroupByException;
import org.apache.carbondata.processing.sort.sortdata.SortParameters;

import org.apache.log4j.Logger;

public class UnsafeInMemoryIntermediateDataMerger implements Callable<Void> {
  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(UnsafeInMemoryIntermediateDataMerger.class.getName());

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

  /**
   * fileCounter
   */
  private int holderCounter;

  /**
   * entryCount
   */
  private int entryCount;

  private UnsafeCarbonRowPage[] unsafeCarbonRowPages;

  private long[] mergedAddresses;

  private byte[] rowPageIndexes;
  private int totalSize;
  private SortParameters sortParameters;
  private SortStepRowHandler sortStepRowHandler;
  private boolean spillDisk;
  private File outputFile;
  private DataOutputStream outputStream;

  /**
   * IntermediateFileMerger Constructor
   */
  public UnsafeInMemoryIntermediateDataMerger(UnsafeCarbonRowPage[] unsafeCarbonRowPages,
      int totalSize, SortParameters sortParameters, boolean spillDisk) {
    this.holderCounter = unsafeCarbonRowPages.length;
    this.unsafeCarbonRowPages = unsafeCarbonRowPages;
    this.mergedAddresses = new long[totalSize];
    this.rowPageIndexes = new byte[totalSize];
    this.entryCount = 0;
    this.totalSize = totalSize;
    this.sortParameters = sortParameters;
    this.sortStepRowHandler = new SortStepRowHandler(sortParameters);
    this.spillDisk = spillDisk;
  }

  @Override
  public Void call() throws Exception {
    long intermediateMergeStartTime = System.currentTimeMillis();
    int holderCounterConst = holderCounter;
    try {
      startSorting();
      if (spillDisk) {
        initSortTempFile();
        while (hasNext()) {
          writeDataToFile(next());
        }
      } else {
        while (hasNext()) {
          writeDataToMemory(next());
        }
      }

      double intermediateMergeCostTime =
          (System.currentTimeMillis() - intermediateMergeStartTime) / 1000.0;
      LOGGER.info("Intermediate Merge of " + holderCounterConst
          + " in-memory sort Cost Time: " + intermediateMergeCostTime + "(s)");
      if (spillDisk) {
        LOGGER.info("Merge and spill in-memory pages to disk, location: "
            + outputFile.getAbsolutePath()
            + ", file size in MB: " + outputFile.length() * 0.1 * 10 / 1024 / 1024
            + ", containing rows: " + totalSize);
      }
    } catch (Exception e) {
      LOGGER.error("Problem while intermediate merging", e);
      throw e;
    } finally {
      if (spillDisk) {
        CarbonUtil.closeStreams(outputStream);
        close();
      }
    }
    return null;
  }

  /**
   * This method will be used to get the sorted record from file
   *
   * @return sorted record sorted record
   * @throws CarbonSortKeyAndGroupByException
   */
  private UnsafeCarbonRowForMerge getSortedRecordFromMemory()
      throws CarbonSortKeyAndGroupByException {
    UnsafeCarbonRowForMerge 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)
    UnsafeInmemoryMergeHolder poll = this.recordHolderHeap.poll();

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

    // check if there no entry present
    if (!poll.hasNext()) {
      // change the file counter
      --this.holderCounter;

      // 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 row pages in intermediate merger: " + this.holderCounter);

    // create record holder heap
    createRecordHolderQueue(unsafeCarbonRowPages);

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

    UnsafeInmemoryMergeHolder unsafePageHolder = null;
    byte index = 0;
    for (UnsafeCarbonRowPage unsafeCarbonRowPage : unsafeCarbonRowPages) {
      // create chunk holder
      unsafePageHolder = new UnsafeInmemoryMergeHolder(unsafeCarbonRowPage, index++);

      // initialize
      unsafePageHolder.readRow();

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

    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
   */
  private void createRecordHolderQueue(UnsafeCarbonRowPage[] pages) {
    // creating record holder heap
    this.recordHolderHeap = new PriorityQueue<UnsafeInmemoryMergeHolder>(pages.length);
  }

  /**
   * This method will be used to get the sorted row
   *
   * @return sorted row
   * @throws CarbonSortKeyAndGroupByException
   */
  private UnsafeCarbonRowForMerge next() throws CarbonSortKeyAndGroupByException {
    if (hasNext()) {
      return getSortedRecordFromMemory();
    } 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.holderCounter > 0;
  }

  /**
   * Below method will be used to write data to file
   */
  private void writeDataToMemory(UnsafeCarbonRowForMerge row) {
    mergedAddresses[entryCount] = row.address;
    rowPageIndexes[entryCount] = row.index;
    entryCount++;
  }

  private void initSortTempFile() throws IOException {
    String tmpDir = sortParameters.getTempFileLocation()[
        new Random().nextInt(sortParameters.getTempFileLocation().length)];
    outputFile = new File(tmpDir + File.separator
        + sortParameters.getTableName() + '_'
        + sortParameters.getRangeId() + '_' + System.nanoTime()
        + CarbonCommonConstants.SORT_TEMP_FILE_EXT);
    outputStream = FileFactory.getDataOutputStream(outputFile.getPath(),
        FileFactory.FileType.LOCAL, sortParameters.getFileWriteBufferSize(),
        sortParameters.getSortTempCompressorName());
    outputStream.writeInt(totalSize);
  }

  private void writeDataToFile(UnsafeCarbonRowForMerge row) throws IOException {
    IntermediateSortTempRow sortTempRow = unsafeCarbonRowPages[row.index].getRow(row.address);
    sortStepRowHandler.writeIntermediateSortTempRowToOutputStream(sortTempRow, outputStream);
  }

  public int getEntryCount() {
    return entryCount;
  }

  public UnsafeCarbonRowPage[] getUnsafeCarbonRowPages() {
    return unsafeCarbonRowPages;
  }

  public long[] getMergedAddresses() {
    return mergedAddresses;
  }

  public byte[] getRowPageIndexes() {
    return rowPageIndexes;
  }

  public boolean isSpillDisk() {
    return spillDisk;
  }

  public void close() {
    for (UnsafeCarbonRowPage rowPage : unsafeCarbonRowPages) {
      rowPage.freeMemory();
    }
  }
}

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

apache / carbondata / 9102

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