public <V> Iterable<V> getAll(TupleTag<V> tag) {
    int index = schema.getIndex(tag);
    if (index < 0) {
      throw new IllegalArgumentException("TupleTag " + tag + " is not in the schema");
    }
    @SuppressWarnings("unchecked")
    Iterable<V> unions = (Iterable<V>) valueMap.get(index);
    return unions;
  }
@Test
  @SuppressWarnings("BoxedPrimitiveEquality")
  public void testCachedResults() {
    // The caching strategies are different for the different implementations.
    Assume.assumeTrue(useReiterator);

    // Ensure we don't fail below due to a non-default java.lang.Integer.IntegerCache.high setting,
    // as we want to test our cache is working as expected, unimpeded by a higher-level cache.
    int integerCacheLimit = 128;
    assertThat(
        Integer.valueOf(integerCacheLimit), not(sameInstance(Integer.valueOf(integerCacheLimit))));

    int perTagCache = 10;
    int crossTagCache = 2 * integerCacheLimit;
    int[] tags = new int[crossTagCache + 8 * perTagCache];
    for (int i = 0; i < 2 * perTagCache; i++) {
      tags[crossTagCache + 4 * i] = 1;
      tags[crossTagCache + 4 * i + 1] = 2;
    }

    TestUnionValues values = new TestUnionValues(tags);
    CoGbkResult result = new CoGbkResult(createSchema(5), values, crossTagCache, perTagCache);

    // More that perTagCache values should be cached for the first tag, as they came first.
    List<Object> tag0 = Lists.newArrayList(result.getAll("tag0").iterator());
    List<Object> tag0again = Lists.newArrayList(result.getAll("tag0").iterator());
    assertThat(tag0.get(0), sameInstance(tag0again.get(0)));
    assertThat(tag0.get(integerCacheLimit), sameInstance(tag0again.get(integerCacheLimit)));
    assertThat(tag0.get(crossTagCache - 1), sameInstance(tag0again.get(crossTagCache - 1)));
    // However, not all elements are cached.
    assertThat(tag0.get(tag0.size() - 1), not(sameInstance(tag0again.get(tag0.size() - 1))));

    // For tag 1 and tag 2, we cache perTagCache elements, plus possibly one more due to peeking
    // iterators.
    List<Object> tag1 = Lists.newArrayList(result.getAll("tag1").iterator());
    List<Object> tag1again = Lists.newArrayList(result.getAll("tag1").iterator());
    assertThat(tag1.get(0), sameInstance(tag1again.get(0)));
    assertThat(tag1.get(perTagCache - 1), sameInstance(tag1again.get(perTagCache - 1)));
    assertThat(tag1.get(perTagCache + 1), not(sameInstance(tag1again.get(perTagCache + 1))));

    List<Object> tag2 = Lists.newArrayList(result.getAll("tag1").iterator());
    List<Object> tag2again = Lists.newArrayList(result.getAll("tag1").iterator());
    assertThat(tag2.get(0), sameInstance(tag2again.get(0)));
    assertThat(tag2.get(perTagCache - 1), sameInstance(tag2again.get(perTagCache - 1)));
    assertThat(tag2.get(perTagCache + 1), not(sameInstance(tag2again.get(perTagCache + 1))));
  }