class FeatureRowDecoder {

  public FeatureRow decode(FeatureRow encodedFeatureRow) {
    if (isEncodedV1(encodedFeatureRow)) {
      // TODO: remove v1 feature row decoder in Feast 0.7
      // Decode Feature Rows using the v1 Decoder.
      final List<Field> fieldsWithoutName = encodedFeatureRow.getFieldsList();
      List<String> featureNames =
          spec.getFeaturesList().stream()
              .sorted(Comparator.comparing(FeatureSpec::getName))
              .map(FeatureSpec::getName)
              .collect(Collectors.toList());

      List<Field> fields =
          IntStream.range(0, featureNames.size())
              .mapToObj(
                  featureNameIndex -> {
                    String featureName = featureNames.get(featureNameIndex);
                    return fieldsWithoutName
                        .get(featureNameIndex)
                        .toBuilder()
                        .setName(featureName)
                        .build();
                  })
              .collect(Collectors.toList());

      return encodedFeatureRow
          .toBuilder()
          .clearFields()
          .setFeatureSet(featureSetRef)
          .addAllFields(fields)
          .build();
    }
    if (isEncodedV2(encodedFeatureRow)) {
      // Decode Feature Rows using the v2 Decoder.
      // v2 Decoder input Feature Rows should use a hashed name as the field name and
      // should not have feature set reference set.
      // Decoding reverts the field name to a unhashed string and set feature set reference.
      Map<String, Value> nameHashValueMap =
          encodedFeatureRow.getFieldsList().stream()
              .collect(Collectors.toMap(field -> field.getName(), field -> field.getValue()));

      List<String> featureNames =
          spec.getFeaturesList().stream().map(FeatureSpec::getName).collect(Collectors.toList());

      List<Field> fields =
          featureNames.stream()
              .map(
                  name -> {
                    String nameHash =
                        Hashing.murmur3_32().hashString(name, StandardCharsets.UTF_8).toString();
                    Value value =
                        nameHashValueMap.getOrDefault(nameHash, Value.newBuilder().build());
                    return Field.newBuilder().setName(name).setValue(value).build();
                  })
              .collect(Collectors.toList());

      return encodedFeatureRow
          .toBuilder()
          .clearFields()
          .setFeatureSet(featureSetRef)
          .addAllFields(fields)
          .build();
    }
    throw new IllegalArgumentException("Failed to decode FeatureRow row: Possible data corruption");
  }

  public  FeatureRowDecoder(String featureSetRef, FeatureSetSpec spec);

}