public List<IntermediateRecord> trimInSegmentResults(GroupKeyGenerator groupKeyGenerator,
      GroupByResultHolder[] groupByResultHolders, int size) {
    // Should not reach here when numGroups <= heap size because there is no need to create a heap
    assert groupKeyGenerator.getNumKeys() > size;
    Iterator<GroupKeyGenerator.GroupKey> groupKeyIterator = groupKeyGenerator.getGroupKeys();
    Comparator<IntermediateRecord> comparator = _intermediateRecordComparator.reversed();

    // Initialize a heap with the first 'size' groups
    IntermediateRecord[] heap = new IntermediateRecord[size];
    for (int i = 0; i < size; i++) {
      heap[i] = getIntermediateRecord(groupKeyIterator.next(), groupByResultHolders);
    }
    makeHeap(heap, size, comparator);

    // Keep updating the heap with the remaining groups
    while (groupKeyIterator.hasNext()) {
      IntermediateRecord intermediateRecord = getIntermediateRecord(groupKeyIterator.next(), groupByResultHolders);
      if (comparator.compare(intermediateRecord, heap[0]) > 0) {
        heap[0] = intermediateRecord;
        downHeap(heap, size, 0, comparator);
      }
    }

    return Arrays.asList(heap);
  }
@Test
  public void testInSegmentTrim() {
    TableResizer tableResizer =
        new TableResizer(DATA_SCHEMA, QueryContextConverterUtils.getQueryContext(QUERY_PREFIX + "d3 DESC"));
    List<IntermediateRecord> results =
        tableResizer.trimInSegmentResults(_groupKeyGenerator, _groupByResultHolders, TRIM_TO_SIZE);
    assertEquals(results.size(), TRIM_TO_SIZE);
    //  _records[4],  _records[3],  _records[2]
    assertEquals(results.get(0)._record, _records.get(2));
    if (results.get(1)._record.equals(_records.get(3))) {
      assertEquals(results.get(2)._record, _records.get(4));
    } else {
      assertEquals(results.get(1)._record, _records.get(4));
      assertEquals(results.get(2)._record, _records.get(3));
    }

    tableResizer = new TableResizer(DATA_SCHEMA, QueryContextConverterUtils.getQueryContext(
        QUERY_PREFIX + "SUM(m1) DESC, max(m2) DESC, DISTINCTCOUNT(m3) DESC"));
    results = tableResizer.trimInSegmentResults(_groupKeyGenerator, _groupByResultHolders, TRIM_TO_SIZE);
    assertEquals(results.size(), TRIM_TO_SIZE);
    // _records[2],  _records[3],  _records[1]
    assertEquals(results.get(0)._record, _records.get(1));
    if (results.get(1)._record.equals(_records.get(3))) {
      assertEquals(results.get(2)._record, _records.get(2));
    } else {
      assertEquals(results.get(1)._record, _records.get(2));
      assertEquals(results.get(2)._record, _records.get(3));
    }

    tableResizer = new TableResizer(DATA_SCHEMA,
        QueryContextConverterUtils.getQueryContext(QUERY_PREFIX + "DISTINCTCOUNT(m3) DESC, AVG(m4) ASC"));
    results = tableResizer.trimInSegmentResults(_groupKeyGenerator, _groupByResultHolders, TRIM_TO_SIZE);
    assertEquals(results.size(), TRIM_TO_SIZE);
    // _records[4],  _records[3],  _records[1]
    assertEquals(results.get(0)._record, _records.get(1));
    if (results.get(1)._record.equals(_records.get(3))) {
      assertEquals(results.get(2)._record, _records.get(4));
    } else {
      assertEquals(results.get(1)._record, _records.get(4));
      assertEquals(results.get(2)._record, _records.get(3));
    }
  }