test_class.declarator: conf
test_class.modifier: private
test_class.original_string: private TajoConf conf;
test_class.type: TajoConf
test_class.var_name: conf
test_class.declarator: util
test_class.modifier: private
test_class.original_string: private TajoTestingCluster util;
test_class.type: TajoTestingCluster
test_class.var_name: util
test_class.declarator: TEST_PATH = "target/test-data/TestExternalSortExec"
test_class.modifier: private final
test_class.original_string: private final String TEST_PATH = "target/test-data/TestExternalSortExec";
test_class.type: String
test_class.var_name: TEST_PATH
test_class.declarator: catalog
test_class.modifier: private
test_class.original_string: private CatalogService catalog;
test_class.type: CatalogService
test_class.var_name: catalog
test_class.declarator: analyzer
test_class.modifier: private
test_class.original_string: private SQLAnalyzer analyzer;
test_class.type: SQLAnalyzer
test_class.var_name: analyzer
test_class.declarator: planner
test_class.modifier: private
test_class.original_string: private LogicalPlanner planner;
test_class.type: LogicalPlanner
test_class.var_name: planner
test_class.declarator: sm
test_class.modifier: private
test_class.original_string: private AbstractStorageManager sm;
test_class.type: AbstractStorageManager
test_class.var_name: sm
test_class.declarator: testDir
test_class.modifier: private
test_class.original_string: private Path testDir;
test_class.type: Path
test_class.var_name: testDir
test_class.declarator: numTuple = 100000
test_class.modifier: private final
test_class.original_string: private final int numTuple = 100000;
test_class.type: int
test_class.var_name: numTuple
test_class.declarator: rnd = new Random(System.currentTimeMillis())
test_class.modifier: private
test_class.original_string: private Random rnd = new Random(System.currentTimeMillis());
test_class.type: Random
test_class.var_name: rnd
test_class.declarator: employee
test_class.modifier: private
test_class.original_string: private TableDesc employee;
test_class.type: TableDesc
test_class.var_name: employee
test_class.declarator: QUERIES = {
        "select managerId, empId from employee order by managerId, empId"
    }
test_class.original_string: String[] QUERIES = {
        "select managerId, empId from employee order by managerId, empId"
    };
test_class.type: String[]
test_class.var_name: QUERIES
test_class.file: tajo-core/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
test_class.identifier: TestExternalSortExec
test_case.body: @Test
    public final void testNext() throws IOException, PlanningException {
      FileFragment[] frags = StorageManager.splitNG(conf, "default.employee", employee.getMeta(), employee.getPath(),
          Integer.MAX_VALUE);
      Path workDir = new Path(testDir, TestExternalSortExec.class.getName());
      TaskAttemptContext ctx = new TaskAttemptContext(conf,
          LocalTajoTestingUtility.newQueryUnitAttemptId(), new FileFragment[] { frags[0] }, workDir);
      ctx.setEnforcer(new Enforcer());
      Expr expr = analyzer.parse(QUERIES[0]);
      LogicalPlan plan = planner.createPlan(LocalTajoTestingUtility.createDummySession(), expr);
      LogicalNode rootNode = plan.getRootBlock().getRoot();
  
      PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf, sm);
      PhysicalExec exec = phyPlanner.createPlan(ctx, rootNode);
      
      ProjectionExec proj = (ProjectionExec) exec;
  
      // TODO - should be planed with user's optimization hint
      if (!(proj.getChild() instanceof ExternalSortExec)) {
        UnaryPhysicalExec sortExec = proj.getChild();
        SeqScanExec scan = sortExec.getChild();
  
        ExternalSortExec extSort = new ExternalSortExec(ctx, sm,
            ((MemSortExec)sortExec).getPlan(), scan);
        proj.setChild(extSort);
      }
  
      Tuple tuple;
      Tuple preVal = null;
      Tuple curVal;
      int cnt = 0;
      exec.init();
      long start = System.currentTimeMillis();
      TupleComparator comparator = new TupleComparator(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 = curVal;
        cnt++;
      }
      long end = System.currentTimeMillis();
      assertEquals(numTuple, cnt);
  
      // for rescan test
      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");
    }
test_case.class_method_signature: TestExternalSortExec.testNext()
test_case.full_signature: @Test public final void testNext()
test_case.identifier: testNext
test_case.invocations: splitNG getMeta getPath getName newQueryUnitAttemptId setEnforcer parse createPlan createDummySession getRoot getRootBlock createPlan getChild getChild getChild getPlan setChild init currentTimeMillis getSchema next assertTrue compare currentTimeMillis assertEquals rescan next assertTrue compare assertEquals close println
test_case.modifiers: @Test public final
test_case.parameters: ()
test_case.return: void
test_case.signature: void testNext()
focal_class.declarator: LOG = LogFactory.getLog(ExternalSortExec.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Log LOG = LogFactory.getLog(ExternalSortExec.class);
focal_class.type: Log
focal_class.var_name: LOG
focal_class.declarator: plan
focal_class.modifier: private
focal_class.original_string: private SortNode plan;
focal_class.type: SortNode
focal_class.var_name: plan
focal_class.declarator: meta
focal_class.modifier: private final
focal_class.original_string: private final TableMeta meta;
focal_class.type: TableMeta
focal_class.var_name: meta
focal_class.declarator: defaultFanout
focal_class.modifier: private final
focal_class.original_string: private final int defaultFanout;
focal_class.type: int
focal_class.var_name: defaultFanout
focal_class.declarator: sortBufferBytesNum
focal_class.modifier: private
focal_class.original_string: private long sortBufferBytesNum;
focal_class.type: long
focal_class.var_name: sortBufferBytesNum
focal_class.declarator: allocatedCoreNum
focal_class.modifier: private final
focal_class.original_string: private final int allocatedCoreNum;
focal_class.type: int
focal_class.var_name: allocatedCoreNum
focal_class.declarator: executorService
focal_class.modifier: private
focal_class.original_string: private ExecutorService executorService;
focal_class.type: ExecutorService
focal_class.var_name: executorService
focal_class.declarator: inMemoryTable
focal_class.modifier: private
focal_class.original_string: private List<Tuple> inMemoryTable;
focal_class.type: List<Tuple>
focal_class.var_name: inMemoryTable
focal_class.declarator: sortTmpDir
focal_class.modifier: private final
focal_class.original_string: private final Path sortTmpDir;
focal_class.type: Path
focal_class.var_name: sortTmpDir
focal_class.declarator: localDirAllocator
focal_class.modifier: private final
focal_class.original_string: private final LocalDirAllocator localDirAllocator;
focal_class.type: LocalDirAllocator
focal_class.var_name: localDirAllocator
focal_class.declarator: localFS
focal_class.modifier: private final
focal_class.original_string: private final RawLocalFileSystem localFS;
focal_class.type: RawLocalFileSystem
focal_class.var_name: localFS
focal_class.declarator: finalOutputFiles = null
focal_class.modifier: private
focal_class.original_string: private List<Path> finalOutputFiles = null;
focal_class.type: List<Path>
focal_class.var_name: finalOutputFiles
focal_class.declarator: mergedInputPaths = null
focal_class.modifier: private
focal_class.original_string: private List<Path> mergedInputPaths = null;
focal_class.type: List<Path>
focal_class.var_name: mergedInputPaths
focal_class.declarator: sorted = false
focal_class.modifier: private
focal_class.original_string: private boolean sorted = false;
focal_class.type: boolean
focal_class.var_name: sorted
focal_class.declarator: memoryResident = true
focal_class.modifier: private
focal_class.original_string: private boolean memoryResident = true;
focal_class.type: boolean
focal_class.var_name: memoryResident
focal_class.declarator: result
focal_class.modifier: private
focal_class.original_string: private Scanner result;
focal_class.type: Scanner
focal_class.var_name: result
focal_class.declarator: sortAndStoredBytes
focal_class.modifier: private
focal_class.original_string: private long sortAndStoredBytes;
focal_class.type: long
focal_class.var_name: sortAndStoredBytes
focal_class.file: tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/ExternalSortExec.java
focal_class.identifier: ExternalSortExec
focal_class.class_method_signature: ExternalSortExec.ExternalSortExec(final TaskAttemptContext context, final AbstractStorageManager sm, final SortNode plan)
focal_class.full_signature: private  ExternalSortExec(final TaskAttemptContext context, final AbstractStorageManager sm, final SortNode plan)
focal_class.identifier: ExternalSortExec
focal_class.modifiers: private
focal_class.parameters: (final TaskAttemptContext context, final AbstractStorageManager sm, final SortNode plan)
focal_class.signature:  ExternalSortExec(final TaskAttemptContext context, final AbstractStorageManager sm, final SortNode plan)
focal_class.class_method_signature: ExternalSortExec.ExternalSortExec(final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan,
                            final CatalogProtos.FragmentProto[] fragments)
focal_class.full_signature: public  ExternalSortExec(final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan,
                            final CatalogProtos.FragmentProto[] fragments)
focal_class.identifier: ExternalSortExec
focal_class.modifiers: public
focal_class.parameters: (final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan,
                            final CatalogProtos.FragmentProto[] fragments)
focal_class.signature:  ExternalSortExec(final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan,
                            final CatalogProtos.FragmentProto[] fragments)
focal_class.class_method_signature: ExternalSortExec.ExternalSortExec(final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan, final PhysicalExec child)
focal_class.full_signature: public  ExternalSortExec(final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan, final PhysicalExec child)
focal_class.identifier: ExternalSortExec
focal_class.modifiers: public
focal_class.parameters: (final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan, final PhysicalExec child)
focal_class.signature:  ExternalSortExec(final TaskAttemptContext context,
                            final AbstractStorageManager sm, final SortNode plan, final PhysicalExec child)
focal_class.class_method_signature: ExternalSortExec.setSortBufferBytesNum(int sortBufferBytesNum)
focal_class.full_signature: @VisibleForTesting public void setSortBufferBytesNum(int sortBufferBytesNum)
focal_class.identifier: setSortBufferBytesNum
focal_class.modifiers: @VisibleForTesting public
focal_class.parameters: (int sortBufferBytesNum)
focal_class.return: void
focal_class.signature: void setSortBufferBytesNum(int sortBufferBytesNum)
focal_class.class_method_signature: ExternalSortExec.init()
focal_class.full_signature: public void init()
focal_class.identifier: init
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void init()
focal_class.class_method_signature: ExternalSortExec.getPlan()
focal_class.full_signature: public SortNode getPlan()
focal_class.identifier: getPlan
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: SortNode
focal_class.signature: SortNode getPlan()
focal_class.class_method_signature: ExternalSortExec.sortAndStoreChunk(int chunkId, List<Tuple> tupleBlock)
focal_class.full_signature: private Path sortAndStoreChunk(int chunkId, List<Tuple> tupleBlock)
focal_class.identifier: sortAndStoreChunk
focal_class.modifiers: private
focal_class.parameters: (int chunkId, List<Tuple> tupleBlock)
focal_class.return: Path
focal_class.signature: Path sortAndStoreChunk(int chunkId, List<Tuple> tupleBlock)
focal_class.class_method_signature: ExternalSortExec.sortAndStoreAllChunks()
focal_class.full_signature: private List<Path> sortAndStoreAllChunks()
focal_class.identifier: sortAndStoreAllChunks
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: List<Path>
focal_class.signature: List<Path> sortAndStoreAllChunks()
focal_class.class_method_signature: ExternalSortExec.getChunkPathForWrite(int level, int chunkId)
focal_class.full_signature: private synchronized Path getChunkPathForWrite(int level, int chunkId)
focal_class.identifier: getChunkPathForWrite
focal_class.modifiers: private synchronized
focal_class.parameters: (int level, int chunkId)
focal_class.return: Path
focal_class.signature: Path getChunkPathForWrite(int level, int chunkId)
focal_class.class_method_signature: ExternalSortExec.next()
focal_class.full_signature: @Override public Tuple next()
focal_class.identifier: next
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: Tuple
focal_class.signature: Tuple next()
focal_class.class_method_signature: ExternalSortExec.calculateFanout(int remainInputChunks, int intputNum, int outputNum, int startIdx)
focal_class.full_signature: private int calculateFanout(int remainInputChunks, int intputNum, int outputNum, int startIdx)
focal_class.identifier: calculateFanout
focal_class.modifiers: private
focal_class.parameters: (int remainInputChunks, int intputNum, int outputNum, int startIdx)
focal_class.return: int
focal_class.signature: int calculateFanout(int remainInputChunks, int intputNum, int outputNum, int startIdx)
focal_class.class_method_signature: ExternalSortExec.externalMergeAndSort(List<Path> chunks)
focal_class.full_signature: private Scanner externalMergeAndSort(List<Path> chunks)
focal_class.identifier: externalMergeAndSort
focal_class.modifiers: private
focal_class.parameters: (List<Path> chunks)
focal_class.return: Scanner
focal_class.signature: Scanner externalMergeAndSort(List<Path> chunks)
focal_class.class_method_signature: ExternalSortExec.checkIfCanBeUnbalancedMerged(int remainInputNum, int outputNum)
focal_class.full_signature: private boolean checkIfCanBeUnbalancedMerged(int remainInputNum, int outputNum)
focal_class.identifier: checkIfCanBeUnbalancedMerged
focal_class.modifiers: private
focal_class.parameters: (int remainInputNum, int outputNum)
focal_class.return: boolean
focal_class.signature: boolean checkIfCanBeUnbalancedMerged(int remainInputNum, int outputNum)
focal_class.class_method_signature: ExternalSortExec.createFinalMerger(List<Path> inputs)
focal_class.full_signature: private Scanner createFinalMerger(List<Path> inputs)
focal_class.identifier: createFinalMerger
focal_class.modifiers: private
focal_class.parameters: (List<Path> inputs)
focal_class.return: Scanner
focal_class.signature: Scanner createFinalMerger(List<Path> inputs)
focal_class.class_method_signature: ExternalSortExec.getFileScanner(Path path)
focal_class.full_signature: private Scanner getFileScanner(Path path)
focal_class.identifier: getFileScanner
focal_class.modifiers: private
focal_class.parameters: (Path path)
focal_class.return: Scanner
focal_class.signature: Scanner getFileScanner(Path path)
focal_class.class_method_signature: ExternalSortExec.createKWayMerger(List<Path> inputs, final int startChunkId, final int num)
focal_class.full_signature: private Scanner createKWayMerger(List<Path> inputs, final int startChunkId, final int num)
focal_class.identifier: createKWayMerger
focal_class.modifiers: private
focal_class.parameters: (List<Path> inputs, final int startChunkId, final int num)
focal_class.return: Scanner
focal_class.signature: Scanner createKWayMerger(List<Path> inputs, final int startChunkId, final int num)
focal_class.class_method_signature: ExternalSortExec.createKWayMergerInternal(final Scanner [] sources, final int startIdx, final int num)
focal_class.full_signature: private Scanner createKWayMergerInternal(final Scanner [] sources, final int startIdx, final int num)
focal_class.identifier: createKWayMergerInternal
focal_class.modifiers: private
focal_class.parameters: (final Scanner [] sources, final int startIdx, final int num)
focal_class.return: Scanner
focal_class.signature: Scanner createKWayMergerInternal(final Scanner [] sources, final int startIdx, final int num)
focal_class.class_method_signature: ExternalSortExec.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: ExternalSortExec.rescan()
focal_class.full_signature: @Override public void rescan()
focal_class.identifier: rescan
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void rescan()
focal_class.class_method_signature: ExternalSortExec.getProgress()
focal_class.full_signature: @Override public float getProgress()
focal_class.identifier: getProgress
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: float
focal_class.signature: float getProgress()
focal_class.class_method_signature: ExternalSortExec.getInputStats()
focal_class.full_signature: @Override public TableStats getInputStats()
focal_class.identifier: getInputStats
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: TableStats
focal_class.signature: TableStats getInputStats()
focal_class.superclass: extends SortExec
focal_method.body: @Override
    public Tuple next() throws IOException {
  
      if (!sorted) { // if not sorted, first sort all data
  
        // if input files are given, it starts merging directly.
        if (mergedInputPaths != null) {
          try {
            this.result = externalMergeAndSort(mergedInputPaths);
          } catch (Exception e) {
            throw new PhysicalPlanningException(e);
          }
        } else {
          // Try to sort all data, and store them as multiple chunks if memory exceeds
          long startTimeOfChunkSplit = System.currentTimeMillis();
          List<Path> chunks = sortAndStoreAllChunks();
          long endTimeOfChunkSplit = System.currentTimeMillis();
          info(LOG, "Chunks creation time: " + (endTimeOfChunkSplit - startTimeOfChunkSplit) + " msec");
  
          if (memoryResident) { // if all sorted data reside in a main-memory table.
            this.result = new MemTableScanner();
          } else { // if input data exceeds main-memory at least once
  
            try {
              this.result = externalMergeAndSort(chunks);
            } catch (Exception e) {
              throw new PhysicalPlanningException(e);
            }
  
          }
        }
  
        sorted = true;
        result.init();
  
        // if loaded and sorted, we assume that it proceeds the half of one entire external sort operation.
        progress = 0.5f;
      }
  
      return result.next();
    }
focal_method.class_method_signature: ExternalSortExec.next()
focal_method.full_signature: @Override public Tuple next()
focal_method.identifier: next
focal_method.invocations: externalMergeAndSort currentTimeMillis sortAndStoreAllChunks currentTimeMillis info externalMergeAndSort init next
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: Tuple
focal_method.signature: Tuple next()