kevinjmh commented on a change in pull request #3603: [CARBONDATA-3679] Optimize local sort performance
URL:
https://github.com/apache/carbondata/pull/3603#discussion_r379267683
##########
File path: processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortDataRows.java
##########
@@ -133,62 +94,74 @@ public void addRow(Object[] row) throws CarbonSortKeyAndGroupByException {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("************ Writing to temp file ********** ");
}
- intermediateFileMerger.startMergingIfPossible();
Object[][] recordHolderListLocal = recordHolderList;
- try {
- semaphore.acquire();
- dataSorterAndWriterExecutorService.execute(new DataSorterAndWriter(recordHolderListLocal));
- } catch (InterruptedException e) {
- LOGGER.error("exception occurred while trying to acquire a semaphore lock: ", e);
- throw new CarbonSortKeyAndGroupByException(e);
- }
+ handlePreviousPage(recordHolderListLocal);
// create the new holder Array
this.recordHolderList = new Object[this.sortBufferSize][];
this.entryCount = 0;
}
recordHolderList[entryCount++] = row;
}
- /**
- * This method will be used to add new row
- *
- * @param rowBatch new rowBatch
- * @throws CarbonSortKeyAndGroupByException problem while writing
- */
public void addRowBatch(Object[][] rowBatch, int size) throws CarbonSortKeyAndGroupByException {
// if record holder list size is equal to sort buffer size then it will
// sort the list and then write current list data to file
- synchronized (addRowsLock) {
- int sizeLeft = 0;
- if (entryCount + size >= sortBufferSize) {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug("************ Writing to temp file ********** ");
- }
- intermediateFileMerger.startMergingIfPossible();
- Object[][] recordHolderListLocal = recordHolderList;
- sizeLeft = sortBufferSize - entryCount;
- if (sizeLeft > 0) {
- System.arraycopy(rowBatch, 0, recordHolderListLocal, entryCount, sizeLeft);
- }
- try {
- semaphore.acquire();
- dataSorterAndWriterExecutorService
- .execute(new DataSorterAndWriter(recordHolderListLocal));
- } catch (Exception e) {
- LOGGER.error(
- "exception occurred while trying to acquire a semaphore lock: " + e.getMessage(), e);
- throw new CarbonSortKeyAndGroupByException(e);
- }
- // create the new holder Array
- this.recordHolderList = new Object[this.sortBufferSize][];
- this.entryCount = 0;
- size = size - sizeLeft;
- if (size == 0) {
- return;
- }
+ int sizeLeft = 0;
+ if (entryCount + size >= sortBufferSize) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("************ Writing to temp file ********** ");
}
- System.arraycopy(rowBatch, sizeLeft, recordHolderList, entryCount, size);
- entryCount += size;
+ Object[][] recordHolderListLocal = recordHolderList;
+ sizeLeft = sortBufferSize - entryCount;
+ if (sizeLeft > 0) {
+ System.arraycopy(rowBatch, 0, recordHolderListLocal, entryCount, sizeLeft);
+ }
+ handlePreviousPage(recordHolderListLocal);
+ // create the new holder Array
+ this.recordHolderList = new Object[this.sortBufferSize][];
+ this.entryCount = 0;
+ size = size - sizeLeft;
+ if (size == 0) {
+ return;
+ }
+ }
+ System.arraycopy(rowBatch, sizeLeft, recordHolderList, entryCount, size);
+ entryCount += size;
+ }
+
+ /**
+ * sort and write data
+ * @param recordHolderArray
+ */
+ private void handlePreviousPage(Object[][] recordHolderArray)
+ throws CarbonSortKeyAndGroupByException {
+ try {
+ long startTime = System.currentTimeMillis();
+ if (parameters.getNumberOfNoDictSortColumns() > 0) {
+ Arrays.sort(recordHolderArray,
+ new NewRowComparator(parameters.getNoDictionarySortColumn(),
+ parameters.getNoDictDataType()));
+ } else {
+ Arrays.sort(recordHolderArray,
+ new NewRowComparatorForNormalDims(parameters.getNumberOfSortColumns()));
+ }
+
+ // create a new file and choose folder randomly every time
+ String[] tmpFileLocation = parameters.getTempFileLocation();
+ String locationChosen = tmpFileLocation[new Random().nextInt(tmpFileLocation.length)];
+ File sortTempFile = new File(
+ locationChosen + File.separator + parameters.getTableName()
+ + '_' + parameters.getRangeId() + '_' + System.nanoTime()
+ + CarbonCommonConstants.SORT_TEMP_FILE_EXT);
+ writeDataToFile(recordHolderArray, recordHolderArray.length, sortTempFile);
+ // add sort temp filename to and arrayList. When the list size reaches 20 then
Review comment:
ok
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[hidden email]
With regards,
Apache Git Services
Locked
