public static <T> Encoder<Tuple2<Integer, T>> oneOfEncoder(List<Encoder<T>> encoders) {
    Expression serializer = serializeOneOf(rootRef(TUPLE2_TYPE, true), encoders);
    Expression deserializer = deserializeOneOf(rootCol(serializer.dataType()), encoders);
    return EncoderFactory.create(serializer, deserializer, Tuple2.class);
  }
@Test
  public void testOneOffEncoder() {
    List<Coder<?>> coders = ImmutableList.copyOf(BASIC_CASES.keySet());
    List<Encoder<?>> encoders = coders.stream().map(EncoderHelpers::encoderFor).collect(toList());

    // build oneOf tuples of type index and corresponding value
    List<Tuple2<Integer, ?>> data =
        BASIC_CASES.entrySet().stream()
            .map(e -> tuple(coders.indexOf(e.getKey()), (Object) e.getValue().get(0)))
            .collect(toList());

    // dataset is a sparse dataset with only one column set per row
    Dataset<Tuple2<Integer, ?>> dataset = createDataset(data, oneOfEncoder((List) encoders));
    assertThat(dataset.collectAsList(), equalTo(data));
  }