Skip to content

Commit 68973c9

Browse files
terryyylimterryyylim
committed
Some cleanups
Signed-off-by: Terence <terencelimxp@gmail.com>
1 parent bad0342 commit 68973c9

3 files changed

Lines changed: 31 additions & 44 deletions

File tree

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/common/RedisHashDecoder.java‎

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

19+
import com.google.common.hash.Hashing;
1920
import com.google.protobuf.InvalidProtocolBufferException;
2021
import com.google.protobuf.Timestamp;
2122
import feast.proto.serving.ServingAPIProto;
2223
import feast.proto.types.ValueProto;
2324
import feast.storage.api.retriever.Feature;
2425
import io.lettuce.core.KeyValue;
26+
import java.nio.charset.StandardCharsets;
2527
import java.util.*;
2628

2729
public class RedisHashDecoder {
@@ -30,14 +32,12 @@ public class RedisHashDecoder {
3032
* Converts all retrieved Redis Hash values based on EntityRows into {@link Feature}
3133
*
3234
* @param redisHashValues retrieved Redis Hash values based on EntityRows
33-
* @param isTimestampMap map to determine if Redis Hash key is a timestamp field
3435
* @param byteToFeatureReferenceMap map to decode bytes back to FeatureReference
3536
* @return List of {@link Feature}
3637
* @throws InvalidProtocolBufferException
3738
*/
3839
public static List<Optional<Feature>> retrieveFeature(
3940
List<KeyValue<byte[], byte[]>> redisHashValues,
40-
Map<String, Boolean> isTimestampMap,
4141
Map<String, ServingAPIProto.FeatureReferenceV2> byteToFeatureReferenceMap,
4242
String timestampPrefix)
4343
throws InvalidProtocolBufferException {
@@ -52,9 +52,9 @@ public static List<Optional<Feature>> retrieveFeature(
5252
byte[] redisValueV = redisHashValues.get(i).getValue();
5353

5454
// Decode data from Redis into Feature object fields
55-
if (isTimestampMap.get(Arrays.toString(redisValueK))) {
55+
if (new String(redisValueK).startsWith(timestampPrefix)) {
5656
Timestamp eventTimestamp = Timestamp.parseFrom(redisValueV);
57-
featureTableTimestampMap.put(Arrays.toString(redisValueK), eventTimestamp);
57+
featureTableTimestampMap.put(new String(redisValueK), eventTimestamp);
5858
} else {
5959
ServingAPIProto.FeatureReferenceV2 featureReference =
6060
byteToFeatureReferenceMap.get(redisValueK.toString());
@@ -68,17 +68,15 @@ public static List<Optional<Feature>> retrieveFeature(
6868
}
6969

7070
// Add timestamp to features
71-
if (allFeaturesBuilderMap.size() > 0) {
72-
for (Map.Entry<ServingAPIProto.FeatureReferenceV2, Optional<Feature.Builder>> entry :
73-
allFeaturesBuilderMap.entrySet()) {
74-
byte[] timestampFeatureTableHashKeyBytes =
75-
RedisHashDecoder.getTimestampRedisHashKeyBytes(entry.getKey(), timestampPrefix);
76-
Timestamp curFeatureTimestamp =
77-
featureTableTimestampMap.get(Arrays.toString(timestampFeatureTableHashKeyBytes));
78-
Feature curFeature = entry.getValue().get().setEventTimestamp(curFeatureTimestamp).build();
79-
allFeatures.add(Optional.of(curFeature));
80-
}
71+
for (Map.Entry<ServingAPIProto.FeatureReferenceV2, Optional<Feature.Builder>> entry :
72+
allFeaturesBuilderMap.entrySet()) {
73+
String timestampRedisHashKeyStr = timestampPrefix + ":" + entry.getKey().getFeatureTable();
74+
Timestamp curFeatureTimestamp = featureTableTimestampMap.get(timestampRedisHashKeyStr);
75+
76+
Feature curFeature = entry.getValue().get().setEventTimestamp(curFeatureTimestamp).build();
77+
allFeatures.add(Optional.of(curFeature));
8178
}
79+
8280
return allFeatures;
8381
}
8482

@@ -87,4 +85,13 @@ public static byte[] getTimestampRedisHashKeyBytes(
8785
String timestampRedisHashKeyStr = timestampPrefix + ":" + featureReference.getFeatureTable();
8886
return timestampRedisHashKeyStr.getBytes();
8987
}
88+
89+
public static byte[] getFeatureReferenceRedisHashKeyBytes(
90+
ServingAPIProto.FeatureReferenceV2 featureReference) {
91+
String delimitedFeatureReference =
92+
featureReference.getFeatureTable() + ":" + featureReference.getName();
93+
return Hashing.murmur3_32()
94+
.hashString(delimitedFeatureReference, StandardCharsets.UTF_8)
95+
.asBytes();
96+
}
9097
}

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

Lines changed: 5 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package feast.storage.connectors.redis.retriever;
1818

1919
import com.google.common.collect.Lists;
20-
import com.google.common.hash.Hashing;
2120
import com.google.protobuf.InvalidProtocolBufferException;
2221
import feast.proto.serving.ServingAPIProto;
2322
import feast.proto.storage.RedisProto;
@@ -35,7 +34,6 @@
3534
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
3635
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
3736
import io.lettuce.core.codec.ByteArrayCodec;
38-
import java.nio.charset.StandardCharsets;
3937
import java.util.*;
4038
import java.util.concurrent.ExecutionException;
4139
import java.util.stream.Collectors;
@@ -75,6 +73,9 @@ private RedisClusterOnlineRetrieverV2(Builder builder) {
7573
this.asyncCommands = builder.connection.async();
7674
this.serializer = builder.serializer;
7775
this.fallbackSerializer = builder.fallbackSerializer;
76+
77+
// Disable auto-flushing
78+
this.asyncCommands.setAutoFlushCommands(false);
7879
}
7980

8081
public static OnlineRetrieverV2 create(Map<String, String> config) {
@@ -154,8 +155,6 @@ private List<List<Optional<Feature>>> getFeaturesFromRedis(
154155
List<List<Optional<Feature>>> features = new ArrayList<>();
155156
// To decode bytes back to Feature Reference
156157
Map<String, ServingAPIProto.FeatureReferenceV2> byteToFeatureReferenceMap = new HashMap<>();
157-
// To check whether redis ValueK is a timestamp field
158-
Map<String, Boolean> isTimestampMap = new HashMap<>();
159158

160159
// Serialize using proto
161160
List<byte[]> binaryRedisKeys =
@@ -167,26 +166,17 @@ private List<List<Optional<Feature>>> getFeaturesFromRedis(
167166
featureReference -> {
168167

169168
// eg. murmur(<featuretable_name:feature_name>)
170-
String delimitedFeatureReference =
171-
featureReference.getFeatureTable() + ":" + featureReference.getName();
172169
byte[] featureReferenceBytes =
173-
Hashing.murmur3_32()
174-
.hashString(delimitedFeatureReference, StandardCharsets.UTF_8)
175-
.asBytes();
170+
RedisHashDecoder.getFeatureReferenceRedisHashKeyBytes(featureReference);
176171
featureReferenceWithTsByteList.add(featureReferenceBytes);
177-
isTimestampMap.put(Arrays.toString(featureReferenceBytes), false);
178172
byteToFeatureReferenceMap.put(featureReferenceBytes.toString(), featureReference);
179173

180174
// eg. <_ts:featuretable_name>
181175
byte[] featureTableTsBytes =
182176
RedisHashDecoder.getTimestampRedisHashKeyBytes(featureReference, timestampPrefix);
183-
isTimestampMap.put(Arrays.toString(featureTableTsBytes), true);
184177
featureReferenceWithTsByteList.add(featureTableTsBytes);
185178
});
186179

187-
// Disable auto-flushing
188-
asyncCommands.setAutoFlushCommands(false);
189-
190180
// Perform a series of independent calls
191181
List<RedisFuture<List<KeyValue<byte[], byte[]>>>> futures = Lists.newArrayList();
192182
for (byte[] binaryRedisKey : binaryRedisKeys) {
@@ -206,7 +196,7 @@ private List<List<Optional<Feature>>> getFeaturesFromRedis(
206196
List<KeyValue<byte[], byte[]>> redisValuesList = future.get();
207197
List<Optional<Feature>> curRedisKeyFeatures =
208198
RedisHashDecoder.retrieveFeature(
209-
redisValuesList, isTimestampMap, byteToFeatureReferenceMap, timestampPrefix);
199+
redisValuesList, byteToFeatureReferenceMap, timestampPrefix);
210200
features.add(curRedisKeyFeatures);
211201
} catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
212202
throw Status.UNKNOWN

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

Lines changed: 5 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package feast.storage.connectors.redis.retriever;
1818

1919
import com.google.common.collect.Lists;
20-
import com.google.common.hash.Hashing;
2120
import com.google.protobuf.InvalidProtocolBufferException;
2221
import feast.proto.serving.ServingAPIProto.FeatureReferenceV2;
2322
import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow;
@@ -34,7 +33,6 @@
3433
import io.lettuce.core.api.StatefulRedisConnection;
3534
import io.lettuce.core.api.async.RedisAsyncCommands;
3635
import io.lettuce.core.codec.ByteArrayCodec;
37-
import java.nio.charset.StandardCharsets;
3836
import java.util.*;
3937
import java.util.concurrent.ExecutionException;
4038
import java.util.stream.Collectors;
@@ -46,6 +44,9 @@ public class RedisOnlineRetrieverV2 implements OnlineRetrieverV2 {
4644

4745
private RedisOnlineRetrieverV2(StatefulRedisConnection<byte[], byte[]> connection) {
4846
this.asyncCommands = connection.async();
47+
48+
// Disable auto-flushing
49+
this.asyncCommands.setAutoFlushCommands(false);
4950
}
5051

5152
public static OnlineRetrieverV2 create(Map<String, String> config) {
@@ -77,8 +78,6 @@ private List<List<Optional<Feature>>> getFeaturesFromRedis(
7778
List<List<Optional<Feature>>> features = new ArrayList<>();
7879
// To decode bytes back to Feature Reference
7980
Map<String, FeatureReferenceV2> byteToFeatureReferenceMap = new HashMap<>();
80-
// To check whether redis ValueK is a timestamp field
81-
Map<String, Boolean> isTimestampMap = new HashMap<>();
8281

8382
// Serialize using proto
8483
List<byte[]> binaryRedisKeys =
@@ -90,26 +89,17 @@ private List<List<Optional<Feature>>> getFeaturesFromRedis(
9089
featureReference -> {
9190

9291
// eg. murmur(<featuretable_name:feature_name>)
93-
String delimitedFeatureReference =
94-
featureReference.getFeatureTable() + ":" + featureReference.getName();
9592
byte[] featureReferenceBytes =
96-
Hashing.murmur3_32()
97-
.hashString(delimitedFeatureReference, StandardCharsets.UTF_8)
98-
.asBytes();
93+
RedisHashDecoder.getFeatureReferenceRedisHashKeyBytes(featureReference);
9994
featureReferenceWithTsByteList.add(featureReferenceBytes);
100-
isTimestampMap.put(Arrays.toString(featureReferenceBytes), false);
10195
byteToFeatureReferenceMap.put(featureReferenceBytes.toString(), featureReference);
10296

10397
// eg. <_ts:featuretable_name>
10498
byte[] featureTableTsBytes =
10599
RedisHashDecoder.getTimestampRedisHashKeyBytes(featureReference, timestampPrefix);
106-
isTimestampMap.put(Arrays.toString(featureTableTsBytes), true);
107100
featureReferenceWithTsByteList.add(featureTableTsBytes);
108101
});
109102

110-
// Disable auto-flushing
111-
asyncCommands.setAutoFlushCommands(false);
112-
113103
// Perform a series of independent calls
114104
List<RedisFuture<List<KeyValue<byte[], byte[]>>>> futures = Lists.newArrayList();
115105
for (byte[] binaryRedisKey : binaryRedisKeys) {
@@ -128,7 +118,7 @@ private List<List<Optional<Feature>>> getFeaturesFromRedis(
128118
List<KeyValue<byte[], byte[]>> redisValuesList = future.get();
129119
List<Optional<Feature>> curRedisKeyFeatures =
130120
RedisHashDecoder.retrieveFeature(
131-
redisValuesList, isTimestampMap, byteToFeatureReferenceMap, timestampPrefix);
121+
redisValuesList, byteToFeatureReferenceMap, timestampPrefix);
132122
features.add(curRedisKeyFeatures);
133123
} catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
134124
throw Status.UNKNOWN

0 commit comments

Comments
 (0)