public static Type beamTypeToSpannerType(Schema.FieldType beamType) {
    switch (beamType.getTypeName()) {
      case ARRAY:
      case ITERABLE:
        Schema.@Nullable FieldType elementType = beamType.getCollectionElementType();
        if (elementType == null) {
          throw new NullPointerException("Null element type");
        } else {
          return Type.array(simpleBeamTypeToSpannerType(elementType));
        }
      default:
        return simpleBeamTypeToSpannerType(beamType);
    }
  }
@Test
  public void testBeamTypeToSpannerTypeTranslation() {
    assertEquals(Type.int64(), beamTypeToSpannerType(Schema.FieldType.INT64));
    assertEquals(Type.int64(), beamTypeToSpannerType(Schema.FieldType.INT32));
    assertEquals(Type.int64(), beamTypeToSpannerType(Schema.FieldType.INT16));
    assertEquals(Type.int64(), beamTypeToSpannerType(Schema.FieldType.BYTE));
    assertEquals(Type.bytes(), beamTypeToSpannerType(Schema.FieldType.BYTES));
    assertEquals(Type.string(), beamTypeToSpannerType(Schema.FieldType.STRING));
    assertEquals(Type.float32(), beamTypeToSpannerType(Schema.FieldType.FLOAT));
    assertEquals(Type.float64(), beamTypeToSpannerType(Schema.FieldType.DOUBLE));
    assertEquals(Type.bool(), beamTypeToSpannerType(Schema.FieldType.BOOLEAN));
    assertEquals(Type.numeric(), beamTypeToSpannerType(Schema.FieldType.DECIMAL));
    assertEquals(
        Type.struct(ImmutableList.of(Type.StructField.of("int64", Type.int64()))),
        beamTypeToSpannerType(Schema.FieldType.row(INT64_SCHEMA)));
    assertEquals(
        Type.array(Type.int64()),
        beamTypeToSpannerType(Schema.FieldType.array(Schema.FieldType.INT64)));
  }