ExternalSortExec extends SortExec { @Override public Tuple next() throws IOException { if (!sorted) { if (mergedInputFragments != null) { try { this.result = externalMergeAndSort(mergedInputFragments); } catch (Exception e) { throw new PhysicalPlanningException(e); } } else { long startTimeOfChunkSplit = System.currentTimeMillis(); List<Path> chunks = sortAndStoreAllChunks(); long endTimeOfChunkSplit = System.currentTimeMillis(); info(LOG, "Chunks creation time: " + (endTimeOfChunkSplit - startTimeOfChunkSplit) + " msec"); if (memoryResident) { TupleSorter sorter = getSorter(inMemoryTable); result = new MemTableScanner(sorter.sort(), inMemoryTable.size(), sortAndStoredBytes); } else { try { List<FileFragment> fragments = TUtil.newList(); for (Path chunk : chunks) { FileFragment frag = new FileFragment("", chunk, 0, new File(localFS.makeQualified(chunk).toUri()).length()); fragments.add(frag); } this.result = externalMergeAndSort(fragments); } catch (Exception e) { throw new PhysicalPlanningException(e); } } } sorted = true; result.init(); progress = 0.5f; } return result.next(); } private  ExternalSortExec(final TaskAttemptContext context, final SortNode plan);  ExternalSortExec(final TaskAttemptContext context,final SortNode plan,
                          final CatalogProtos.FragmentProto[] fragments);  ExternalSortExec(final TaskAttemptContext context, final SortNode plan, final PhysicalExec child); @VisibleForTesting void setSortBufferBytesNum(int sortBufferBytesNum); void init(); SortNode getPlan(); @Override Tuple next(); @Override void close(); @Override void rescan(); @Override float getProgress(); @Override TableStats getInputStats();  }
@Test public final void testNext() throws IOException, PlanningException { FileFragment[] frags = FileStorageManager.splitNG(conf, "default.employee", employee.getMeta(), new Path(employee.getPath()), Integer.MAX_VALUE); Path workDir = new Path(testDir, TestExternalSortExec.class.getName()); TaskAttemptContext ctx = new TaskAttemptContext(new QueryContext(conf), LocalTajoTestingUtility.newTaskAttemptId(), new FileFragment[] { frags[0] }, workDir); ctx.setEnforcer(new Enforcer()); Expr expr = analyzer.parse(QUERIES[0]); LogicalPlan plan = planner.createPlan(LocalTajoTestingUtility.createDummyContext(conf), expr); LogicalNode rootNode = plan.getRootBlock().getRoot(); PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf); PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode); ProjectionExec proj = (ProjectionExec) exec; ExternalSortExec extSort; if (!(proj.getChild() instanceof ExternalSortExec)) { UnaryPhysicalExec sortExec = proj.getChild(); SeqScanExec scan = sortExec.getChild(); extSort = new ExternalSortExec(ctx, ((MemSortExec)sortExec).getPlan(), scan); proj.setChild(extSort); } else { extSort = proj.getChild(); } extSort.setSortBufferBytesNum(1024*1024); Tuple tuple; Tuple preVal = null; Tuple curVal; int cnt = 0; exec.init(); long start = System.currentTimeMillis(); BaseTupleComparator comparator = new BaseTupleComparator(proj.getSchema(), new SortSpec[]{ new SortSpec(new Column("managerid", Type.INT4)), new SortSpec(new Column("empid", Type.INT4)) }); while ((tuple = exec.next()) != null) { curVal = tuple; if (preVal != null) { assertTrue("prev: " + preVal + ", but cur: " + curVal, comparator.compare(preVal, curVal) <= 0); } preVal = new VTuple(curVal); cnt++; } long end = System.currentTimeMillis(); assertEquals(numTuple, cnt); preVal = null; exec.rescan(); cnt = 0; while ((tuple = exec.next()) != null) { curVal = tuple; if (preVal != null) { assertTrue("prev: " + preVal + ", but cur: " + curVal, comparator.compare(preVal, curVal) <= 0); } preVal = curVal; cnt++; } assertEquals(numTuple, cnt); exec.close(); System.out.println("Sort Time: " + (end - start) + " msc"); }