@Override
  public void finish(boolean sort, boolean storeFinalResult) {
    if (_hasOrderBy) {
      long startTimeNs = System.nanoTime();
      _topRecords = _tableResizer.getTopRecords(_lookupMap, _resultSize, sort);
      long resizeTimeNs = System.nanoTime() - startTimeNs;
      _numResizes++;
      _resizeTimeNs += resizeTimeNs;
    } else {
      _topRecords = _lookupMap.values();
    }
    // TODO: Directly return final result in _tableResizer.getTopRecords to avoid extracting final result multiple times
    assert !(_hasFinalInput && !storeFinalResult);
    if (storeFinalResult && !_hasFinalInput) {
      ColumnDataType[] columnDataTypes = _dataSchema.getColumnDataTypes();
      int numAggregationFunctions = _aggregationFunctions.length;
      for (int i = 0; i < numAggregationFunctions; i++) {
        columnDataTypes[i + _numKeyColumns] = _aggregationFunctions[i].getFinalResultColumnType();
      }
      for (Record record : _topRecords) {
        Object[] values = record.getValues();
        for (int i = 0; i < numAggregationFunctions; i++) {
          int colId = i + _numKeyColumns;
          values[colId] = _aggregationFunctions[i].extractFinalResult(values[colId]);
        }
      }
    }
  }
@Test(dataProvider = "initDataProvider")
  public void testNonConcurrentIndexedTable(String orderBy, List<String> survivors) {
    QueryContext queryContext = QueryContextConverterUtils.getQueryContext(
        "SELECT SUM(m1), MAX(m2) FROM testTable GROUP BY d1, d2, d3, d4 ORDER BY " + orderBy);
    DataSchema dataSchema =
        new DataSchema(new String[]{"d1", "d2", "d3", "d4", "sum(m1)", "max(m2)"}, new ColumnDataType[]{
            ColumnDataType.STRING, ColumnDataType.INT, ColumnDataType.DOUBLE, ColumnDataType.INT,
            ColumnDataType.DOUBLE, ColumnDataType.DOUBLE
        });

    // Test SimpleIndexedTable
    IndexedTable indexedTable = new SimpleIndexedTable(dataSchema, queryContext, 5, TRIM_SIZE, TRIM_THRESHOLD);
    IndexedTable mergeTable = new SimpleIndexedTable(dataSchema, queryContext, 10, TRIM_SIZE, TRIM_THRESHOLD);
    testNonConcurrent(indexedTable, mergeTable);
    indexedTable.finish(true);
    checkSurvivors(indexedTable, survivors);

    // Test ConcurrentIndexedTable
    indexedTable = new ConcurrentIndexedTable(dataSchema, queryContext, 5, TRIM_SIZE, TRIM_THRESHOLD);
    mergeTable = new SimpleIndexedTable(dataSchema, queryContext, 10, TRIM_SIZE, TRIM_THRESHOLD);
    testNonConcurrent(indexedTable, mergeTable);
    indexedTable.finish(true);
    checkSurvivors(indexedTable, survivors);
  }