shunlean commented on a change in pull request #3847:
URL:
https://github.com/apache/carbondata/pull/3847#discussion_r460741367##########
File path: processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/UnsafeSortDataRows.java
##########
@@ -200,25 +203,44 @@ public void startSorting() {
* @param file file
* @throws CarbonSortKeyAndGroupByException
*/
- private void writeDataToFile(UnsafeCarbonRowPage rowPage, File file)
- throws CarbonSortKeyAndGroupByException {
- DataOutputStream stream = null;
- try {
- // open stream
- stream = FileFactory.getDataOutputStream(file.getPath(),
- parameters.getFileWriteBufferSize(), parameters.getSortTempCompressorName());
- int actualSize = rowPage.getBuffer().getActualSize();
- // write number of entries to the file
- stream.writeInt(actualSize);
- for (int i = 0; i < actualSize; i++) {
- rowPage.writeRow(
- rowPage.getBuffer().get(i) + rowPage.getDataBlock().getBaseOffset(), stream);
+ private void writeDataToFile(UnsafeCarbonRowPage rowPage, File file) {
+ writeService.submit(new WriteThread(rowPage, file));
+ }
+
+ public class WriteThread implements Runnable {
+ private File file;
+ private UnsafeCarbonRowPage rowPage;
+
+ public WriteThread(UnsafeCarbonRowPage rowPage, File file) {
+ this.rowPage = rowPage;
+ this.file = file;
+
+ }
+
+ @Override
+ public void run() {
+ DataOutputStream stream = null;
+ try {
+ // open stream
+ stream = FileFactory.getDataOutputStream(this.file.getPath(),
+ parameters.getFileWriteBufferSize(), parameters.getSortTempCompressorName());
+ int actualSize = rowPage.getBuffer().getActualSize();
+ // write number of entries to the file
+ stream.writeInt(actualSize);
+ for (int i = 0; i < actualSize; i++) {
+ rowPage.writeRow(
+ rowPage.getBuffer().get(i) + rowPage.getDataBlock().getBaseOffset(), stream);
+ }
+ // add sort temp filename to and arrayList. When the list size reaches 20 then
+ // intermediate merging of sort temp files will be triggered
+ unsafeInMemoryIntermediateFileMerger.addFileToMerge(file);
+ } catch (IOException | MemoryException e) {
+ e.printStackTrace();
Review comment:
ok, done.
##########
File path: processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortParameters.java
##########
@@ -37,6 +40,13 @@
import org.apache.log4j.Logger;
public class SortParameters implements Serializable {
+
+ private ExecutorService writeService = Executors.newFixedThreadPool(5,
Review comment:
ok,done.
----------------------------------------------------------------
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]
Locked
