@Override
  public ParsedSchema toParsedSchema(final PersistenceSchema schema) {
    SerdeUtils.throwOnUnsupportedFeatures(schema.features(), format.supportedFeatures());

    final ConnectSchema outerSchema = ConnectSchemas.columnsToConnectSchema(schema.columns());
    final ConnectSchema innerSchema = SerdeUtils
        .applySinglesUnwrapping(outerSchema, schema.features());

    return connectSrTranslator.fromConnectSchema(innerSchema);
  }
@Test
  public void shouldSupportBuildingPrimitiveSchemas() {
    // Given:
    when(format.supportedFeatures()).thenReturn(ImmutableSet.of(SerdeFeature.UNWRAP_SINGLES));


    // When:
    translator.toParsedSchema(
        PersistenceSchema.from(
            ImmutableList.of(createColumn("bob", SqlTypes.INTEGER)),
            SerdeFeatures.of(SerdeFeature.UNWRAP_SINGLES)
        )
    );

    // Then:
    verify(innerTranslator).fromConnectSchema(SchemaBuilder.int32().optional().build());
  }