Skip to content

Commit ae29ba9

Browse files
mrzzyOleksii Moskalenko
mrzzy
authored and
Oleksii Moskalenko
committed
Fix Online Serving unable to retrieve feature data after Feature Set update. (#908)
* Update RedisCustomIO to write FeatureRows with field's name set to hash of field. * Update FeatureRowDecoder to decode by name hash instead of order * Bump pytest order numbers by 2 to make space for new tests * Revert "Bump pytest order numbers by 2 to make space for new tests" This reverts commit aecc9a6e9a70be3fd84d04f81442b518be01a4c6. * Added e2e to check that feature rows with missing or extra fields can be retrieved * Clarify docs about Feature Row v1 encoding and Feature Row v2 encoding * Fix python lint * Update FeatureRowDecoder's isEncodedV2 check to use anyMatch() * Make missing field/extra field e2e tests independent of other tests. * Update FeatureRowDecoder if/else statement into 2 ifs * Fix python and java lint * Fix java unit test failures * Fix ImportJobTest java unit test * Sync github workflows with master * Sync .github folder with master for fix * Replace v1/v2 encoding with v1/v2 decoder in docs
1 parent a466f64 commit ae29ba9

11 files changed

Lines changed: 404 additions & 96 deletions

File tree

‎ingestion/src/test/java/feast/ingestion/ImportJobTest.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -217,7 +217,9 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow()
217217
.map(FeatureSpec::getName)
218218
.collect(Collectors.toList())
219219
.contains(field.getName()))
220-
.map(field -> field.toBuilder().clearName().build())
220+
.map(
221+
field ->
222+
field.toBuilder().setName(TestUtil.hash(field.getName())).build())
221223
.collect(Collectors.toList());
222224
randomRow =
223225
randomRow

‎ingestion/src/test/java/feast/test/TestUtil.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import static feast.common.models.FeatureSet.getFeatureSetStringRef;
2020

2121
import com.google.common.collect.ImmutableList;
22+
import com.google.common.hash.Hashing;
2223
import com.google.common.io.Files;
2324
import com.google.protobuf.ByteString;
2425
import com.google.protobuf.Message;
@@ -517,4 +518,8 @@ public static void waitUntilAllElementsAreWrittenToStore(
517518
}
518519
}
519520
}
521+
522+
public static String hash(String input) {
523+
return Hashing.murmur3_32().hashString(input, StandardCharsets.UTF_8).toString();
524+
}
520525
}

‎storage/api/src/main/java/feast/storage/common/testing/TestUtil.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
*/
1717
package feast.storage.common.testing;
1818

19+
import com.google.common.hash.Hashing;
1920
import com.google.protobuf.ByteString;
2021
import com.google.protobuf.Timestamp;
2122
import feast.proto.core.FeatureSetProto.FeatureSet;
@@ -24,6 +25,7 @@
2425
import feast.proto.types.FeatureRowProto.FeatureRow.Builder;
2526
import feast.proto.types.FieldProto.Field;
2627
import feast.proto.types.ValueProto.*;
28+
import java.nio.charset.StandardCharsets;
2729
import java.time.Instant;
2830
import java.util.concurrent.ThreadLocalRandom;
2931
import org.apache.commons.lang3.RandomStringUtils;
@@ -191,4 +193,8 @@ public static Field field(String name, Object value, ValueType.Enum valueType) {
191193
throw new IllegalStateException("Unexpected valueType: " + value.getClass());
192194
}
193195
}
196+
197+
public static String hash(String input) {
198+
return Hashing.murmur3_32().hashString(input, StandardCharsets.UTF_8).toString();
199+
}
194200
}

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/retriever/FeatureRowDecoder.java‎

Lines changed: 81 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,17 @@
1616
*/
1717
package feast.storage.connectors.redis.retriever;
1818

19+
import com.google.common.hash.Hashing;
1920
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
2021
import feast.proto.core.FeatureSetProto.FeatureSpec;
2122
import feast.proto.types.FeatureRowProto.FeatureRow;
2223
import feast.proto.types.FieldProto.Field;
24+
import feast.proto.types.ValueProto.Value;
25+
import feast.storage.connectors.redis.writer.RedisCustomIO;
26+
import java.nio.charset.StandardCharsets;
2327
import java.util.Comparator;
2428
import java.util.List;
29+
import java.util.Map;
2530
import java.util.stream.Collectors;
2631
import java.util.stream.IntStream;
2732

@@ -36,60 +41,102 @@ public FeatureRowDecoder(String featureSetRef, FeatureSetSpec spec) {
3641
}
3742

3843
/**
39-
* A feature row is considered encoded if the feature set and field names are not set. This method
40-
* is required for backward compatibility purposes, to allow Feast serving to continue serving non
41-
* encoded Feature Row ingested by an older version of Feast.
44+
* Check if encoded feature row can be decoded by v1 Decoder. The v1 Decoder requires that the
45+
* Feature Row to have both it's feature set reference and fields names are not set. The no. of
46+
* fields in the feature row should also match up with the number of fields in the Feature Set
47+
* spec. NOTE: This method is deprecated and will be removed in Feast v0.7.
4248
*
4349
* @param featureRow Feature row
4450
* @return boolean
4551
*/
46-
public boolean isEncoded(FeatureRow featureRow) {
52+
@Deprecated
53+
private boolean isEncodedV1(FeatureRow featureRow) {
4754
return featureRow.getFeatureSet().isEmpty()
48-
&& featureRow.getFieldsList().stream().allMatch(field -> field.getName().isEmpty());
55+
&& featureRow.getFieldsList().stream().allMatch(field -> field.getName().isEmpty())
56+
&& featureRow.getFieldsList().size() == spec.getFeaturesList().size();
4957
}
5058

5159
/**
52-
* Validates if an encoded feature row can be decoded without exception.
60+
* Check if encoded feature row can be decoded by Decoder. The v2 Decoder requires that a Feature
61+
* Row to have both it feature set reference and fields names are set.
5362
*
5463
* @param featureRow Feature row
5564
* @return boolean
5665
*/
57-
public boolean isEncodingValid(FeatureRow featureRow) {
58-
return featureRow.getFieldsList().size() == spec.getFeaturesList().size();
66+
private boolean isEncodedV2(FeatureRow featureRow) {
67+
return !featureRow.getFieldsList().stream().anyMatch(field -> field.getName().isEmpty());
5968
}
6069

6170
/**
62-
* Decoding feature row by repopulating the field names based on the corresponding feature set
63-
* spec.
71+
* Decode feature row encoded by {@link RedisCustomIO}. NOTE: The v1 Decoder will be removed in
72+
* Feast 0.7
6473
*
74+
* @throws IllegalArgumentException if unable to the decode the given feature row
6575
* @param encodedFeatureRow Feature row
6676
* @return boolean
6777
*/
6878
public FeatureRow decode(FeatureRow encodedFeatureRow) {
69-
final List<Field> fieldsWithoutName = encodedFeatureRow.getFieldsList();
79+
if (isEncodedV1(encodedFeatureRow)) {
80+
// TODO: remove v1 feature row decoder in Feast 0.7
81+
// Decode Feature Rows using the v1 Decoder.
82+
final List<Field> fieldsWithoutName = encodedFeatureRow.getFieldsList();
83+
List<String> featureNames =
84+
spec.getFeaturesList().stream()
85+
.sorted(Comparator.comparing(FeatureSpec::getName))
86+
.map(FeatureSpec::getName)
87+
.collect(Collectors.toList());
7088

71-
List<String> featureNames =
72-
spec.getFeaturesList().stream()
73-
.sorted(Comparator.comparing(FeatureSpec::getName))
74-
.map(FeatureSpec::getName)
75-
.collect(Collectors.toList());
76-
List<Field> fields =
77-
IntStream.range(0, featureNames.size())
78-
.mapToObj(
79-
featureNameIndex -> {
80-
String featureName = featureNames.get(featureNameIndex);
81-
return fieldsWithoutName
82-
.get(featureNameIndex)
83-
.toBuilder()
84-
.setName(featureName)
85-
.build();
86-
})
87-
.collect(Collectors.toList());
88-
return encodedFeatureRow
89-
.toBuilder()
90-
.clearFields()
91-
.setFeatureSet(featureSetRef)
92-
.addAllFields(fields)
93-
.build();
89+
List<Field> fields =
90+
IntStream.range(0, featureNames.size())
91+
.mapToObj(
92+
featureNameIndex -> {
93+
String featureName = featureNames.get(featureNameIndex);
94+
return fieldsWithoutName
95+
.get(featureNameIndex)
96+
.toBuilder()
97+
.setName(featureName)
98+
.build();
99+
})
100+
.collect(Collectors.toList());
101+
102+
return encodedFeatureRow
103+
.toBuilder()
104+
.clearFields()
105+
.setFeatureSet(featureSetRef)
106+
.addAllFields(fields)
107+
.build();
108+
}
109+
if (isEncodedV2(encodedFeatureRow)) {
110+
// Decode Feature Rows using the v2 Decoder.
111+
// v2 Decoder input Feature Rows should use a hashed name as the field name and
112+
// should not have feature set reference set.
113+
// Decoding reverts the field name to a unhashed string and set feature set reference.
114+
Map<String, Value> nameHashValueMap =
115+
encodedFeatureRow.getFieldsList().stream()
116+
.collect(Collectors.toMap(field -> field.getName(), field -> field.getValue()));
117+
118+
List<String> featureNames =
119+
spec.getFeaturesList().stream().map(FeatureSpec::getName).collect(Collectors.toList());
120+
121+
List<Field> fields =
122+
featureNames.stream()
123+
.map(
124+
name -> {
125+
String nameHash =
126+
Hashing.murmur3_32().hashString(name, StandardCharsets.UTF_8).toString();
127+
Value value =
128+
nameHashValueMap.getOrDefault(nameHash, Value.newBuilder().build());
129+
return Field.newBuilder().setName(name).setValue(value).build();
130+
})
131+
.collect(Collectors.toList());
132+
133+
return encodedFeatureRow
134+
.toBuilder()
135+
.clearFields()
136+
.setFeatureSet(featureSetRef)
137+
.addAllFields(fields)
138+
.build();
139+
}
140+
throw new IllegalArgumentException("Failed to decode FeatureRow row: Possible data corruption");
94141
}
95142
}

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisClusterOnlineRetriever.java‎

Lines changed: 5 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -158,17 +158,11 @@ private List<Optional<FeatureRow>> getFeaturesFromRedis(
158158

159159
// decode feature rows from data bytes using decoder.
160160
FeatureRow featureRow = FeatureRow.parseFrom(featureRowBytes);
161-
if (decoder.isEncoded(featureRow)) {
162-
if (decoder.isEncodingValid(featureRow)) {
163-
featureRow = decoder.decode(featureRow);
164-
} else {
165-
// decoding feature row failed: data corruption could have occurred
166-
throw Status.DATA_LOSS
167-
.withDescription(
168-
"Failed to decode FeatureRow from bytes retrieved from redis"
169-
+ ": Possible data corruption")
170-
.asRuntimeException();
171-
}
161+
try {
162+
featureRow = decoder.decode(featureRow);
163+
} catch (IllegalArgumentException e) {
164+
// decoding feature row failed: data corruption could have occurred
165+
throw Status.DATA_LOSS.withCause(e).withDescription(e.getMessage()).asRuntimeException();
172166
}
173167
featureRows.add(Optional.of(featureRow));
174168
}

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisOnlineRetriever.java‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -151,15 +151,11 @@ private List<Optional<FeatureRow>> getFeaturesFromRedis(
151151

152152
// decode feature rows from data bytes using decoder.
153153
FeatureRow featureRow = FeatureRow.parseFrom(featureRowBytes);
154-
if (decoder.isEncoded(featureRow) && decoder.isEncodingValid(featureRow)) {
154+
try {
155155
featureRow = decoder.decode(featureRow);
156-
} else {
156+
} catch (IllegalArgumentException e) {
157157
// decoding feature row failed: data corruption could have occurred
158-
throw Status.DATA_LOSS
159-
.withDescription(
160-
"Failed to decode FeatureRow from bytes retrieved from redis"
161-
+ ": Possible data corruption")
162-
.asRuntimeException();
158+
throw Status.DATA_LOSS.withCause(e).withDescription(e.getMessage()).asRuntimeException();
163159
}
164160
featureRows.add(Optional.of(featureRow));
165161
}

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java‎

Lines changed: 29 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
import com.google.common.collect.Iterators;
2020
import com.google.common.collect.Lists;
21+
import com.google.common.hash.Hashing;
2122
import feast.proto.core.FeatureSetProto.EntitySpec;
2223
import feast.proto.core.FeatureSetProto.FeatureSetSpec;
2324
import feast.proto.core.FeatureSetProto.FeatureSpec;
@@ -29,7 +30,9 @@
2930
import feast.storage.api.writer.FailedElement;
3031
import feast.storage.api.writer.WriteResult;
3132
import feast.storage.common.retry.Retriable;
33+
import feast.storage.connectors.redis.retriever.FeatureRowDecoder;
3234
import io.lettuce.core.RedisException;
35+
import java.nio.charset.StandardCharsets;
3336
import java.util.HashMap;
3437
import java.util.List;
3538
import java.util.Map;
@@ -203,28 +206,45 @@ private byte[] getKey(FeatureRow featureRow, FeatureSetSpec spec) {
203206
return redisKeyBuilder.build().toByteArray();
204207
}
205208

209+
/**
210+
* Encode the Feature Row as bytes to store in Redis in encoded Feature Row encoding. To
211+
* reduce storage space consumption in redis, feature rows are "encoded" by hashing the fields
212+
* names and not unsetting the feature set reference. {@link FeatureRowDecoder} is
213+
* rensponsible for reversing this "encoding" step.
214+
*/
206215
private byte[] getValue(FeatureRow featureRow, FeatureSetSpec spec) {
207216
List<String> featureNames =
208217
spec.getFeaturesList().stream().map(FeatureSpec::getName).collect(Collectors.toList());
209-
Map<String, Field> fieldValueOnlyMap =
218+
219+
Map<String, Field.Builder> fieldValueOnlyMap =
210220
featureRow.getFieldsList().stream()
211221
.filter(field -> featureNames.contains(field.getName()))
212222
.distinct()
213223
.collect(
214224
Collectors.toMap(
215-
Field::getName,
216-
field -> Field.newBuilder().setValue(field.getValue()).build()));
225+
Field::getName, field -> Field.newBuilder().setValue(field.getValue())));
217226

218227
List<Field> values =
219228
featureNames.stream()
220229
.sorted()
221230
.map(
222-
featureName ->
223-
fieldValueOnlyMap.getOrDefault(
224-
featureName,
225-
Field.newBuilder()
226-
.setValue(ValueProto.Value.getDefaultInstance())
227-
.build()))
231+
featureName -> {
232+
Field.Builder field =
233+
fieldValueOnlyMap.getOrDefault(
234+
featureName,
235+
Field.newBuilder().setValue(ValueProto.Value.getDefaultInstance()));
236+
237+
// Encode the name of the as the hash of the field name.
238+
// Use hash of name instead of the name of to reduce redis storage consumption
239+
// per feature row stored.
240+
String nameHash =
241+
Hashing.murmur3_32()
242+
.hashString(featureName, StandardCharsets.UTF_8)
243+
.toString();
244+
field.setName(nameHash);
245+
246+
return field.build();
247+
})
228248
.collect(Collectors.toList());
229249

230250
return FeatureRow.newBuilder()

0 commit comments

Comments
 (0)