@VisibleForTesting
  static Object scalarToProtoValue(FieldType beamFieldType, Object value) {
    if (beamFieldType.getTypeName() == TypeName.LOGICAL_TYPE) {
      @Nullable LogicalType<?, ?> logicalType = beamFieldType.getLogicalType();
      if (logicalType == null) {
        throw new RuntimeException("Unexpectedly null logical type " + beamFieldType);
      }
      @Nullable
      BiFunction<LogicalType<?, ?>, Object, Object> logicalTypeEncoder =
          LOGICAL_TYPE_ENCODERS.get(logicalType.getIdentifier());
      if (logicalTypeEncoder == null) {
        throw new RuntimeException("Unsupported logical type " + logicalType.getIdentifier());
      }
      return logicalTypeEncoder.apply(logicalType, value);
    } else {
      @Nullable
      Function<Object, Object> encoder = PRIMITIVE_ENCODERS.get(beamFieldType.getTypeName());
      if (encoder == null) {
        throw new RuntimeException("Unexpected beam type " + beamFieldType);
      }
      return encoder.apply(value);
    }
  }
@Test
  public void testScalarToProtoValue() {
    Map<FieldType, Iterable<Pair<Object, Object>>> testCases =
        ImmutableMap.<FieldType, Iterable<Pair<Object, Object>>>builder()
            .put(
                FieldType.BYTES,
                ImmutableList.of(
                    Pair.create(BYTES, ByteString.copyFrom(BYTES)),
                    Pair.create(ByteBuffer.wrap(BYTES), ByteString.copyFrom(BYTES)),
                    Pair.create(
                        new String(BYTES, StandardCharsets.UTF_8), ByteString.copyFrom(BYTES))))
            .build();
    for (Map.Entry<FieldType, Iterable<Pair<Object, Object>>> entry : testCases.entrySet()) {
      entry
          .getValue()
          .forEach(
              p -> {
                assertEquals(
                    p.getValue(),
                    BeamRowToStorageApiProto.scalarToProtoValue(entry.getKey(), p.getKey()));
              });
    }
  }