PrecombineGroupingTable implements GroupingTable<K, InputT, AccumT> { @VisibleForTesting public long size() { return size; } PrecombineGroupingTable(
      long maxSize,
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      Combiner<? super K, InputT, AccumT, ?> combineFn,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super AccumT> accumulatorSizer); static GroupingTable<WindowedValue<K>, InputT, AccumT> combining(
      PipelineOptions options,
      CombineFn<InputT, AccumT, ?> combineFn,
      Coder<K> keyCoder,
      Coder<? super AccumT> accumulatorCoder); static GroupingTable<WindowedValue<K>, InputT, AccumT> combiningAndSampling(
          PipelineOptions options,
          CombineFn<InputT, AccumT, ?> combineFn,
          Coder<K> keyCoder,
          Coder<? super AccumT> accumulatorCoder,
          double sizeEstimatorSampleRate); @SuppressWarnings("unchecked") @Override void put(Object pair, Receiver receiver); void put(K key, InputT value, Receiver receiver); @Override void flush(Receiver output); @VisibleForTesting void setMaxSize(long maxSize); @VisibleForTesting long size();  }
@Test public void testSampleHighVarianceSizes() throws Exception { List<Long> sizes = Arrays.asList(1L, 10L, 100L, 1000L); IdentitySizeEstimator underlying = new IdentitySizeEstimator(); SizeEstimator<Long> estimator = new SamplingSizeEstimator<>(underlying, 0.1, 0.2, 10, new Random(1)); for (int k = 0; k < 10; k++) { long size = sizes.get(k % sizes.size()); assertEquals(size, estimator.estimateSize(size)); assertEquals(k + 1, underlying.calls); } for (int k = 10; k < 20; k++) { long size = sizes.get(k % sizes.size()); assertEquals(size, estimator.estimateSize(size)); assertEquals(k + 1, underlying.calls); } for (int k = 20; k < 500; k++) { estimator.estimateSize(sizes.get(k % sizes.size())); } int initialCalls = underlying.calls; for (int k = 500; k < 1500; k++) { long size = sizes.get(k % sizes.size()); assertThat(estimator.estimateSize(size), anyOf(is(in(sizes)), between(250L, 350L))); } assertThat(underlying.calls - initialCalls, between(180, 220)); for (int k = 1500; k < 3000; k++) { estimator.estimateSize(sizes.get(k % sizes.size())); } initialCalls = underlying.calls; for (int k = 3000; k < 4000; k++) { long size = sizes.get(k % sizes.size()); assertThat(estimator.estimateSize(size), anyOf(is(in(sizes)), between(250L, 350L))); } assertThat(underlying.calls - initialCalls, between(90, 110)); }