FeatureRowDecoder { public FeatureRow decode(FeatureRow encodedFeatureRow) { if (isEncodedV1(encodedFeatureRow)) { 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)) { 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"); } FeatureRowDecoder(String featureSetRef, FeatureSetSpec spec); FeatureRow decode(FeatureRow encodedFeatureRow);  }
@Test public void shouldDecodeValidEncodedFeatureRowV2() { FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); FeatureRowProto.FeatureRow encodedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder() .setName( Hashing.murmur3_32() .hashString("feature1", StandardCharsets.UTF_8) .toString()) .setValue(Value.newBuilder().setInt32Val(2))) .addFields( Field.newBuilder() .setName( Hashing.murmur3_32() .hashString("feature2", StandardCharsets.UTF_8) .toString()) .setValue(Value.newBuilder().setFloatVal(1.0f))) .build(); FeatureRowProto.FeatureRow expectedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setFeatureSet("feature_set_ref") .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder().setName("feature1").setValue(Value.newBuilder().setInt32Val(2))) .addFields( Field.newBuilder() .setName("feature2") .setValue(Value.newBuilder().setFloatVal(1.0f))) .build(); assertEquals(expectedFeatureRow, decoder.decode(encodedFeatureRow)); }
@Test public void shouldDecodeValidFeatureRowV2WithIncompleteFields() { FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); FeatureRowProto.FeatureRow encodedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder() .setName( Hashing.murmur3_32() .hashString("feature1", StandardCharsets.UTF_8) .toString()) .setValue(Value.newBuilder().setInt32Val(2))) .build(); FeatureRowProto.FeatureRow expectedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setFeatureSet("feature_set_ref") .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder().setName("feature1").setValue(Value.newBuilder().setInt32Val(2))) .addFields(Field.newBuilder().setName("feature2").setValue(Value.newBuilder().build())) .build(); assertEquals(expectedFeatureRow, decoder.decode(encodedFeatureRow)); }
@Test public void shouldDecodeValidFeatureRowV2AndIgnoreExtraFields() { FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); FeatureRowProto.FeatureRow encodedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder() .setName( Hashing.murmur3_32() .hashString("feature1", StandardCharsets.UTF_8) .toString()) .setValue(Value.newBuilder().setInt32Val(2))) .addFields( Field.newBuilder() .setName( Hashing.murmur3_32() .hashString("feature2", StandardCharsets.UTF_8) .toString()) .setValue(Value.newBuilder().setFloatVal(1.0f))) .addFields( Field.newBuilder() .setName( Hashing.murmur3_32() .hashString("feature3", StandardCharsets.UTF_8) .toString()) .setValue(Value.newBuilder().setStringVal("data"))) .build(); FeatureRowProto.FeatureRow expectedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setFeatureSet("feature_set_ref") .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder().setName("feature1").setValue(Value.newBuilder().setInt32Val(2))) .addFields( Field.newBuilder() .setName("feature2") .setValue(Value.newBuilder().setFloatVal(1.0f))) .build(); assertEquals(expectedFeatureRow, decoder.decode(encodedFeatureRow)); }
@Test public void shouldDecodeValidEncodedFeatureRowV1() { FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); FeatureRowProto.FeatureRow encodedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields(Field.newBuilder().setValue(Value.newBuilder().setInt32Val(2))) .addFields(Field.newBuilder().setValue(Value.newBuilder().setFloatVal(1.0f))) .build(); FeatureRowProto.FeatureRow expectedFeatureRow = FeatureRowProto.FeatureRow.newBuilder() .setFeatureSet("feature_set_ref") .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) .addFields( Field.newBuilder().setName("feature1").setValue(Value.newBuilder().setInt32Val(2))) .addFields( Field.newBuilder() .setName("feature2") .setValue(Value.newBuilder().setFloatVal(1.0f))) .build(); assertEquals(expectedFeatureRow, decoder.decode(encodedFeatureRow)); }