GroupingTables { public static <K, InputT, AccumT> GroupingTable<K, InputT, AccumT> combining( GroupingKeyCreator<? super K> groupingKeyCreator, PairInfo pairInfo, Combiner<? super K, InputT, AccumT, ?> combineFn, SizeEstimator<? super K> keySizer, SizeEstimator<? super AccumT> accumulatorSizer) { return new CombiningGroupingTable<>( DEFAULT_MAX_GROUPING_TABLE_BYTES, groupingKeyCreator, pairInfo, combineFn, keySizer, accumulatorSizer); }  static GroupingTable<K, V, List<V>> buffering(
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super V> valueSizer); static GroupingTable<K, V, List<V>> buffering(
      Long maxTableSizeBytes,
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super V> valueSizer); static GroupingTable<K, V, List<V>> bufferingAndSampling(
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super V> valueSizer,
      double sizeEstimatorSampleRate,
      long maxSizeBytes); static GroupingTable<K, InputT, AccumT> combining(
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      Combiner<? super K, InputT, AccumT, ?> combineFn,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super AccumT> accumulatorSizer); static GroupingTable<K, InputT, AccumT> combiningAndSampling(
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      Combiner<? super K, InputT, AccumT, ?> combineFn,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super AccumT> accumulatorSizer,
      double sizeEstimatorSampleRate,
      long maxSizeBytes);  }
@Test public void testCombiningGroupingTable() throws Exception { Combiner<Object, Integer, Long, Long> summingCombineFn = new Combiner<Object, Integer, Long, Long>() { @Override public Long createAccumulator(Object key) { return 0L; } @Override public Long add(Object key, Long accumulator, Integer value) { return accumulator + value; } @Override public Long merge(Object key, Iterable<Long> accumulators) { long sum = 0; for (Long part : accumulators) { sum += part; } return sum; } @Override public Long compact(Object key, Long accumulator) { return accumulator; } @Override public Long extract(Object key, Long accumulator) { return accumulator; } }; GroupingTableBase<String, Integer, Long> table = (GroupingTableBase<String, Integer, Long>) GroupingTables.combining( new IdentityGroupingKeyCreator(), new KvPairInfo(), summingCombineFn, new StringPowerSizeEstimator(), new IdentitySizeEstimator()); table.setMaxSize(1000); TestOutputReceiver receiver = new TestOutputReceiver( KvCoder.of(StringUtf8Coder.of(), BigEndianLongCoder.of()), NameContextsForTests.nameContextForTest()); table.put("A", 1, receiver); table.put("B", 2, receiver); table.put("B", 3, receiver); table.put("C", 4, receiver); assertThat(receiver.outputElems, empty()); table.put("C", 5000, receiver); assertThat(receiver.outputElems, hasItem((Object) KV.of("C", 5004L))); table.put("DDDD", 6, receiver); assertThat(receiver.outputElems, hasItem((Object) KV.of("DDDD", 6L))); table.flush(receiver); assertThat( receiver.outputElems, IsIterableContainingInAnyOrder.<Object>containsInAnyOrder( KV.of("A", 1L), KV.of("B", 2L + 3), KV.of("C", 5000L + 4), KV.of("DDDD", 6L))); }