public static <T> ReadAll<T> readAll() {
    return new AutoValue_CassandraIO_ReadAll.Builder<T>().build();
  }
@Test
  public void testReadAllRingRange() {
    RingRange physRR =
        fromEncodedKey(
            cluster.getMetadata(), TypeCodec.varchar().serialize("phys", ProtocolVersion.V3));

    RingRange mathRR =
        fromEncodedKey(
            cluster.getMetadata(), TypeCodec.varchar().serialize("math", ProtocolVersion.V3));

    RingRange logicRR =
        fromEncodedKey(
            cluster.getMetadata(), TypeCodec.varchar().serialize("logic", ProtocolVersion.V3));

    PCollection<Scientist> output =
        pipeline
            .apply(Create.of(getReadWithRingRange(physRR), getReadWithRingRange(mathRR, logicRR)))
            .apply(
                CassandraIO.<Scientist>readAll().withCoder(SerializableCoder.of(Scientist.class)));

    PCollection<KV<String, Integer>> mapped =
        output.apply(
            MapElements.via(
                new SimpleFunction<Scientist, KV<String, Integer>>() {
                  @Override
                  public KV<String, Integer> apply(Scientist scientist) {
                    return KV.of(scientist.department, scientist.id);
                  }
                }));

    PAssert.that(mapped.apply("Count occurrences per department", Count.perKey()))
        .satisfies(
            input -> {
              HashMap<String, Long> map = new HashMap<>();
              for (KV<String, Long> element : input) {
                map.put(element.getKey(), element.getValue());
              }
              assertEquals(3, map.size()); // do we have all three departments
              assertEquals(10L, (long) map.get("phys"));
              assertEquals(4L, (long) map.get("math"));
              assertEquals(2L, (long) map.get("logic"));
              return null;
            });

    pipeline.run();
  }