@Test
  public void testMemoryPressureFlush() throws HiveException {

    List<String> mapColumnNames = new ArrayList<String>();
    mapColumnNames.add("Key");
    mapColumnNames.add("Value");
    VectorizationContext ctx = new VectorizationContext("name", mapColumnNames);

    Pair<GroupByDesc,VectorGroupByDesc> pair = buildKeyGroupByDesc (ctx, "max",
        "Value", TypeInfoFactory.longTypeInfo,
        new String[] {"Key"},
        new TypeInfo[] {TypeInfoFactory.longTypeInfo});
    GroupByDesc desc = pair.left;
    VectorGroupByDesc vectorDesc = pair.right;

    // Set the memory threshold so that we get 100Kb before we need to flush.
    MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
    long maxMemory = memoryMXBean.getHeapMemoryUsage().getMax();

    float threshold = 100.0f*1024.0f/maxMemory;
    desc.setGroupByMemoryUsage(threshold);

    CompilationOpContext cCtx = new CompilationOpContext();

    Operator<? extends OperatorDesc> groupByOp = OperatorFactory.get(cCtx, desc);

    VectorGroupByOperator vgo =
        (VectorGroupByOperator) Vectorizer.vectorizeGroupByOperator(groupByOp, ctx, vectorDesc);

    FakeCaptureVectorToRowOutputOperator out = FakeCaptureVectorToRowOutputOperator.addCaptureOutputChild(cCtx, vgo);
    vgo.initialize(hconf, null);

    long expected = vgo.getMaxMemory();
    assertEquals(expected, maxMemory);
    this.outputRowCount = 0;
    out.setOutputInspector(new FakeCaptureVectorToRowOutputOperator.OutputInspector() {
      @Override
      public void inspectRow(Object row, int tag) throws HiveException {
        ++outputRowCount;
      }
    });

    Iterable<Object> it = new Iterable<Object>() {
      @Override
      public Iterator<Object> iterator() {
        return new Iterator<Object> () {
          long value = 0;

          @Override
          public boolean hasNext() {
            return true;
          }

          @Override
          public Object next() {
            return ++value;
          }

          @Override
          public void remove() {
          }
        };
      }
    };

    FakeVectorRowBatchFromObjectIterables data = new FakeVectorRowBatchFromObjectIterables(
        100,
        new String[] {"long", "long"},
        it,
        it);

    // The 'it' data source will produce data w/o ever ending
    // We want to see that memory pressure kicks in and some
    // entries in the VGBY are flushed.
    long countRowsProduced = 0;
    for (VectorizedRowBatch unit: data) {
      countRowsProduced += 100;
      vgo.process(unit,  0);
      if (0 < outputRowCount) {
        break;
      }
      // Set an upper bound how much we're willing to push before it should flush
      // we've set the memory threshold at 100kb, each key is distinct
      // It should not go beyond 100k/16 (key+data)
      assertTrue(countRowsProduced < 100*1024/16);
    }

    assertTrue(0 < outputRowCount);
  }
Unit test for the vectorized GROUP BY operator.
ql/src/test/org/apache/hadoop/hive/ql/exec/vector/TestVectorGroupByOperator.java