From be192bd6e620df51ec4e5ff12ffdbf7fd2b72539 Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 14 Oct 2020 15:50:03 +0800 Subject: [PATCH 1/7] Update java sdk Signed-off-by: Terence --- .../java/com/gojek/feast/FeastClient.java | 63 ++++--------------- .../java/com/gojek/feast/RequestUtil.java | 38 ++++++----- .../java/com/gojek/feast/FeastClientTest.java | 50 +++++++-------- .../java/com/gojek/feast/RequestUtilTest.java | 50 +++++++++------ 4 files changed, 89 insertions(+), 112 deletions(-) diff --git a/sdk/java/src/main/java/com/gojek/feast/FeastClient.java b/sdk/java/src/main/java/com/gojek/feast/FeastClient.java index 4cd056f31d4..f8414b4429c 100644 --- a/sdk/java/src/main/java/com/gojek/feast/FeastClient.java +++ b/sdk/java/src/main/java/com/gojek/feast/FeastClient.java @@ -16,11 +16,11 @@ */ package com.gojek.feast; -import feast.proto.serving.ServingAPIProto.FeatureReference; +import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; import feast.proto.serving.ServingAPIProto.GetFeastServingInfoRequest; import feast.proto.serving.ServingAPIProto.GetFeastServingInfoResponse; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequest; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequest.EntityRow; +import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2; +import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow; import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesResponse; import feast.proto.serving.ServingServiceGrpc; import feast.proto.serving.ServingServiceGrpc.ServingServiceBlockingStub; @@ -102,50 +102,17 @@ public static FeastClient createSecure(String host, int port, SecurityConfig sec /** * Obtain info about Feast Serving. * - * @return {@link feast.proto.serving.ServingAPIProto.GetFeastServingInfoResponse} containing - * Feast version, Serving type etc. + * @return {@link GetFeastServingInfoResponse} containing Feast version, Serving type etc. */ public GetFeastServingInfoResponse getFeastServingInfo() { return stub.getFeastServingInfo(GetFeastServingInfoRequest.newBuilder().build()); } - /** - * Get online features from Feast from FeatureSets - * - *

See {@link #getOnlineFeatures(List, List, String, boolean)} - * - * @param featureRefs list of string feature references to retrieve in the following format - * featureSet:feature, where 'featureSet' and 'feature' refer to the FeatureSet and Feature - * names respectively. Only the Feature name is required. - * @param rows list of {@link Row} to select the entities to retrieve the features for. - * @return list of {@link Row} containing retrieved data fields. - */ - public List getOnlineFeatures(List featureRefs, List rows) { - return getOnlineFeatures(featureRefs, rows, ""); - } - - /** - * Get online features from Feast. - * - *

See {@link #getOnlineFeatures(List, List, String, boolean)} - * - * @param featureRefs list of string feature references to retrieve in the following format - * featureSet:feature, where 'featureSet' and 'feature' refer to the FeatureSet and Feature - * names respectively. Only the Feature name is required. - * @param rows list of {@link Row} to select the entities to retrieve the features for - * @param project {@link String} Specifies the project override. If specifed uses the project for - * retrieval. Overrides the projects set in Feature References if also specified. - * @return list of {@link Row} containing retrieved data fields. - */ - public List getOnlineFeatures(List featureRefs, List rows, String project) { - return getOnlineFeatures(featureRefs, rows, project, false); - } - /** * Get online features from Feast. * - *

Example of retrieving online features for the driver featureset, with features driver_id and - * driver_name + *

Example of retrieving online features for the driver FeatureTable, with features driver_id + * and driver_name * * 

{@code
    * FeastClient client = FeastClient.create("localhost", 6566);
@@ -157,18 +124,15 @@ public List getOnlineFeatures(List featureRefs, List rows, Str
    * }
* * @param featureRefs list of string feature references to retrieve in the following format - * featureSet:feature, where 'featureSet' and 'feature' refer to the FeatureSet and Feature - * names respectively. Only the Feature name is required. + * featureTable:feature, where 'featureTable' and 'feature' refer to the FeatureTable and + * Feature names respectively. Only the Feature name is required. * @param rows list of {@link Row} to select the entities to retrieve the features for * @param project {@link String} Specifies the project override. If specifed uses the project for * retrieval. Overrides the projects set in Feature References if also specified. - * @param omitEntitiesInResponse if true, the returned {@link Row} will not contain field and - * value for the entity * @return list of {@link Row} containing retrieved data fields. */ - public List getOnlineFeatures( - List featureRefs, List rows, String project, boolean omitEntitiesInResponse) { - List features = RequestUtil.createFeatureRefs(featureRefs); + public List getOnlineFeatures(List featureRefs, List rows, String project) { + List features = RequestUtil.createFeatureRefs(featureRefs); // build entity rows and collect entity references HashSet entityRefs = new HashSet<>(); List entityRows = @@ -177,19 +141,18 @@ public List getOnlineFeatures( row -> { entityRefs.addAll(row.getFields().keySet()); return EntityRow.newBuilder() - .setEntityTimestamp(row.getEntityTimestamp()) + .setTimestamp(row.getEntityTimestamp()) .putAllFields(row.getFields()) .build(); }) .collect(Collectors.toList()); GetOnlineFeaturesResponse response = - stub.getOnlineFeatures( - GetOnlineFeaturesRequest.newBuilder() + stub.getOnlineFeaturesV2( + GetOnlineFeaturesRequestV2.newBuilder() .addAllFeatures(features) .addAllEntityRows(entityRows) .setProject(project) - .setOmitEntitiesInResponse(omitEntitiesInResponse) .build()); return response.getFieldValuesList().stream() diff --git a/sdk/java/src/main/java/com/gojek/feast/RequestUtil.java b/sdk/java/src/main/java/com/gojek/feast/RequestUtil.java index c2290c6d11d..69c8f9f737a 100644 --- a/sdk/java/src/main/java/com/gojek/feast/RequestUtil.java +++ b/sdk/java/src/main/java/com/gojek/feast/RequestUtil.java @@ -16,7 +16,7 @@ */ package com.gojek.feast; -import feast.proto.serving.ServingAPIProto.FeatureReference; +import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; import java.util.List; import java.util.stream.Collectors; @@ -27,28 +27,28 @@ public class RequestUtil { * Create feature references protos from given string feature reference. * * @param featureRefStrings to create Feature Reference protos from - * @return List of parsed {@link FeatureReference} protos + * @return List of parsed {@link FeatureReferenceV2} protos */ - public static List createFeatureRefs(List featureRefStrings) { + public static List createFeatureRefs(List featureRefStrings) { if (featureRefStrings == null) { - throw new IllegalArgumentException("featureRefs cannot be null"); + throw new IllegalArgumentException("FeatureReferences cannot be null"); } - List featureRefs = + List featureRefs = featureRefStrings.stream() .map(refStr -> parseFeatureRef(refStr)) .collect(Collectors.toList()); - return featureRefs.stream().map(ref -> ref.build()).collect(Collectors.toList()); + return featureRefs; } /** * Parse a feature reference proto builder from the given featureRefString * * @param featureRefString string feature reference to parse from. - * @return a parsed {@link FeatureReference.Builder} + * @return a parsed {@link FeatureReferenceV2} */ - public static FeatureReference.Builder parseFeatureRef(String featureRefString) { + public static FeatureReferenceV2 parseFeatureRef(String featureRefString) { featureRefString = featureRefString.trim(); if (featureRefString.isEmpty()) { throw new IllegalArgumentException("Cannot parse a empty feature reference"); @@ -60,15 +60,21 @@ public static FeatureReference.Builder parseFeatureRef(String featureRefString) + " Feature References is not longer supported: %s", featureRefString)); } - - FeatureReference.Builder featureRef = FeatureReference.newBuilder(); - // parse featureset if specified - if (featureRefString.contains(":")) { - String[] featureSetSplit = featureRefString.split(":"); - featureRef.setFeatureSet(featureSetSplit[0]); - featureRefString = featureSetSplit[1]; + if (!featureRefString.contains(":")) { + throw new IllegalArgumentException( + String.format( + "Unsupported feature reference: %s - FeatureTable name and Feature name should be provided in string" + + " Feature References, in : format.", + featureRefString)); } - featureRef.setName(featureRefString); + + String[] featureReferenceParts = featureRefString.split(":"); + FeatureReferenceV2 featureRef = + FeatureReferenceV2.newBuilder() + .setFeatureTable(featureReferenceParts[0]) + .setName(featureReferenceParts[1]) + .build(); + return featureRef; } } diff --git a/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java b/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java index 0c583be98c4..c458a06425f 100644 --- a/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java +++ b/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java @@ -22,21 +22,16 @@ import com.google.protobuf.Timestamp; import feast.common.auth.credentials.JwtCallCredentials; -import feast.proto.serving.ServingAPIProto.FeatureReference; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequest; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequest.EntityRow; +import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; +import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2; +import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow; import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesResponse; import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesResponse.FieldStatus; import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesResponse.FieldValues; import feast.proto.serving.ServingServiceGrpc.ServingServiceImplBase; import feast.proto.types.ValueProto.Value; -import io.grpc.ManagedChannel; -import io.grpc.Metadata; -import io.grpc.ServerCall; +import io.grpc.*; import io.grpc.ServerCall.Listener; -import io.grpc.ServerCallHandler; -import io.grpc.ServerInterceptor; -import io.grpc.Status; import io.grpc.inprocess.InProcessChannelBuilder; import io.grpc.inprocess.InProcessServerBuilder; import io.grpc.stub.StreamObserver; @@ -63,8 +58,8 @@ public class FeastClientTest { delegatesTo( new ServingServiceImplBase() { @Override - public void getOnlineFeatures( - GetOnlineFeaturesRequest request, + public void getOnlineFeaturesV2( + GetOnlineFeaturesRequestV2 request, StreamObserver responseObserver) { if (!request.equals(FeastClientTest.getFakeRequest())) { responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); @@ -132,7 +127,7 @@ public void shouldAuthenticateAndGetOnlineFeatures() { private void shouldGetOnlineFeaturesWithClient(FeastClient client) { List rows = client.getOnlineFeatures( - Arrays.asList("driver:name", "rating", "null_value"), + Arrays.asList("driver:name", "driver:rating", "driver:null_value"), Arrays.asList( Row.create().set("driver_id", 1).setEntityTimestamp(Instant.ofEpochSecond(100))), "driver_project"); @@ -143,8 +138,8 @@ private void shouldGetOnlineFeaturesWithClient(FeastClient client) { { put("driver_id", intValue(1)); put("driver:name", strValue("david")); - put("rating", intValue(3)); - put("null_value", Value.newBuilder().build()); + put("driver:rating", intValue(3)); + put("driver:null_value", Value.newBuilder().build()); } }); assertEquals( @@ -153,21 +148,24 @@ private void shouldGetOnlineFeaturesWithClient(FeastClient client) { { put("driver_id", FieldStatus.PRESENT); put("driver:name", FieldStatus.PRESENT); - put("rating", FieldStatus.PRESENT); - put("null_value", FieldStatus.NULL_VALUE); + put("driver:rating", FieldStatus.PRESENT); + put("driver:null_value", FieldStatus.NULL_VALUE); } }); } - private static GetOnlineFeaturesRequest getFakeRequest() { + private static GetOnlineFeaturesRequestV2 getFakeRequest() { // setup mock serving service stub - return GetOnlineFeaturesRequest.newBuilder() - .addFeatures(FeatureReference.newBuilder().setFeatureSet("driver").setName("name").build()) - .addFeatures(FeatureReference.newBuilder().setName("rating").build()) - .addFeatures(FeatureReference.newBuilder().setName("null_value").build()) + return GetOnlineFeaturesRequestV2.newBuilder() + .addFeatures( + FeatureReferenceV2.newBuilder().setFeatureTable("driver").setName("name").build()) + .addFeatures( + FeatureReferenceV2.newBuilder().setFeatureTable("driver").setName("rating").build()) + .addFeatures( + FeatureReferenceV2.newBuilder().setFeatureTable("driver").setName("null_value").build()) .addEntityRows( EntityRow.newBuilder() - .setEntityTimestamp(Timestamp.newBuilder().setSeconds(100)) + .setTimestamp(Timestamp.newBuilder().setSeconds(100)) .putFields("driver_id", intValue(1))) .setProject("driver_project") .build(); @@ -181,10 +179,10 @@ private static GetOnlineFeaturesResponse getFakeResponse() { .putStatuses("driver_id", FieldStatus.PRESENT) .putFields("driver:name", strValue("david")) .putStatuses("driver:name", FieldStatus.PRESENT) - .putFields("rating", intValue(3)) - .putStatuses("rating", FieldStatus.PRESENT) - .putFields("null_value", Value.newBuilder().build()) - .putStatuses("null_value", FieldStatus.NULL_VALUE) + .putFields("driver:rating", intValue(3)) + .putStatuses("driver:rating", FieldStatus.PRESENT) + .putFields("driver:null_value", Value.newBuilder().build()) + .putStatuses("driver:null_value", FieldStatus.NULL_VALUE) .build()) .build(); } diff --git a/sdk/java/src/test/java/com/gojek/feast/RequestUtilTest.java b/sdk/java/src/test/java/com/gojek/feast/RequestUtilTest.java index 1148710d5a6..1592e20664d 100644 --- a/sdk/java/src/test/java/com/gojek/feast/RequestUtilTest.java +++ b/sdk/java/src/test/java/com/gojek/feast/RequestUtilTest.java @@ -21,8 +21,8 @@ import com.google.common.collect.ImmutableList; import com.google.protobuf.TextFormat; -import feast.common.models.Feature; -import feast.proto.serving.ServingAPIProto.FeatureReference; +import feast.common.models.FeatureV2; +import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; import java.util.Arrays; import java.util.Comparator; import java.util.List; @@ -38,20 +38,22 @@ class RequestUtilTest { private static Stream provideValidFeatureRefs() { return Stream.of( Arguments.of( - Arrays.asList("driver:driver_id", "driver_id"), + Arrays.asList("driver:driver_id"), Arrays.asList( - FeatureReference.newBuilder().setFeatureSet("driver").setName("driver_id").build(), - FeatureReference.newBuilder().setName("driver_id").build()))); + FeatureReferenceV2.newBuilder() + .setFeatureTable("driver") + .setName("driver_id") + .build()))); } @ParameterizedTest @MethodSource("provideValidFeatureRefs") - void createFeatureSets_ShouldReturnFeatureSetsForValidFeatureRefs( - List input, List expected) { - List actual = RequestUtil.createFeatureRefs(input); - // Order of the actual and expected featureSets do no not matter - actual.sort(Comparator.comparing(FeatureReference::getName)); - expected.sort(Comparator.comparing(FeatureReference::getName)); + void createFeatureRefs_ShouldReturnFeaturesForValidFeatureRefs( + List input, List expected) { + List actual = RequestUtil.createFeatureRefs(input); + // Order of the actual and expected FeatureTables do no not matter + actual.sort(Comparator.comparing(FeatureReferenceV2::getName)); + expected.sort(Comparator.comparing(FeatureReferenceV2::getName)); assertEquals(expected.size(), actual.size()); for (int i = 0; i < expected.size(); i++) { String expectedString = TextFormat.printer().printToString(expected.get(i)); @@ -63,13 +65,10 @@ void createFeatureSets_ShouldReturnFeatureSetsForValidFeatureRefs( @ParameterizedTest @MethodSource("provideValidFeatureRefs") void renderFeatureRef_ShouldReturnFeatureRefString( - List expected, List input) { - input = - input.stream() - .map(ref -> ref.toBuilder().clearProject().build()) - .collect(Collectors.toList()); + List expected, List input) { + input = input.stream().map(ref -> ref.toBuilder().build()).collect(Collectors.toList()); List actual = - input.stream().map(ref -> Feature.getFeatureStringRef(ref)).collect(Collectors.toList()); + input.stream().map(ref -> FeatureV2.getFeatureStringRef(ref)).collect(Collectors.toList()); assertEquals(expected.size(), actual.size()); for (int i = 0; i < expected.size(); i++) { assertEquals(expected.get(i), actual.get(i)); @@ -77,18 +76,29 @@ void renderFeatureRef_ShouldReturnFeatureRefString( } private static Stream provideInvalidFeatureRefs() { - return Stream.of(Arguments.of(ImmutableList.of("project/feature", ""))); + return Stream.of(Arguments.of(ImmutableList.of("project/feature"))); + } + + private static Stream provideMissingFeatureTableFeatureRefs() { + return Stream.of(Arguments.of(ImmutableList.of("feature"))); } @ParameterizedTest @MethodSource("provideInvalidFeatureRefs") - void createFeatureSets_ShouldThrowExceptionForInvalidFeatureRefs(List input) { + void createFeatureRefs_ShouldThrowExceptionForProjectInFeatureRefs(List input) { + assertThrows(IllegalArgumentException.class, () -> RequestUtil.createFeatureRefs(input)); + } + + @ParameterizedTest + @MethodSource("provideMissingFeatureTableFeatureRefs") + void createFeatureRefs_ShouldThrowExceptionForMissingFeatureTableInFeatureRefs( + List input) { assertThrows(IllegalArgumentException.class, () -> RequestUtil.createFeatureRefs(input)); } @ParameterizedTest @NullSource - void createFeatureSets_ShouldThrowExceptionForNullFeatureRefs(List input) { + void createFeatureRefs_ShouldThrowExceptionForNullFeatureRefs(List input) { assertThrows(IllegalArgumentException.class, () -> RequestUtil.createFeatureRefs(input)); } } From 72b8f5ab7ced7630fc6e91b1e4d99d91ebf147ee Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 14 Oct 2020 16:05:26 +0800 Subject: [PATCH 2/7] Update python sdk Signed-off-by: Terence --- sdk/python/feast/client.py | 120 +++++++++++++++++++++++++++++++- sdk/python/feast/feature.py | 52 ++++++++++++++ sdk/python/tests/test_client.py | 118 ++++++++++++++++++++++++++----- 3 files changed, 269 insertions(+), 21 deletions(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index d023795ae5f..d012396598f 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -14,7 +14,7 @@ import logging import multiprocessing import shutil -from typing import Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Union, cast import grpc import pandas as pd @@ -56,6 +56,7 @@ from feast.core.CoreService_pb2_grpc import CoreServiceStub from feast.data_source import BigQuerySource, FileSource from feast.entity import Entity +from feast.feature import FeatureRef from feast.feature_table import FeatureTable from feast.grpc import auth as feast_auth from feast.grpc.grpc import create_grpc_channel @@ -68,8 +69,15 @@ _write_non_partitioned_table_from_source, _write_partitioned_table_from_source, ) -from feast.serving.ServingService_pb2 import GetFeastServingInfoRequest +from feast.online_response import OnlineResponse +from feast.serving.ServingService_pb2 import ( + FeatureReferenceV2, + GetFeastServingInfoRequest, + GetOnlineFeaturesRequestV2, +) from feast.serving.ServingService_pb2_grpc import ServingServiceStub +from feast.type_map import _python_value_to_proto_value, python_type_to_feast_value_type +from feast.types.Value_pb2 import Value as Value _logger = logging.getLogger(__name__) @@ -709,3 +717,111 @@ def _get_grpc_metadata(self): if self._config.getboolean(CONFIG_ENABLE_AUTH_KEY) and self._auth_metadata: return self._auth_metadata.get_signed_meta() return () + + def get_online_features( + self, + feature_refs: List[str], + entity_rows: List[Dict[str, Any]], + project: Optional[str] = None, + ) -> OnlineResponse: + """ + Retrieves the latest online feature data from Feast Serving. + + Args: + feature_refs: List of feature references that will be returned for each entity. + Each feature reference should have the following format: + "feature_table:feature" where "feature_table" & "feature" refer to + the feature and feature table names respectively. + Only the feature name is required. + entity_rows: A list of dictionaries where each key-value is an entity-name, entity-value pair. + project: Optionally specify the the project override. If specified, uses given project for retrieval. + Overrides the projects specified in Feature References if also are specified. + + Returns: + GetOnlineFeaturesResponse containing the feature data in records. + Each EntityRow provided will yield one record, which contains + data fields with data value and field status metadata (if included). + + Examples: + >>> from feast import Client + >>> + >>> feast_client = Client(core_url="localhost:6565", serving_url="localhost:6566") + >>> feature_refs = ["sales:daily_transactions"] + >>> entity_rows = [{"customer_id": 0},{"customer_id": 1}] + >>> + >>> online_response = feast_client.get_online_features( + >>> feature_refs, entity_rows, project="my_project") + >>> online_response_dict = online_response.to_dict() + >>> print(online_response_dict) + {'sales:daily_transactions': [1.1,1.2], 'sales:customer_id': [0,1]} + """ + + try: + response = self._serving_service.GetOnlineFeaturesV2( + GetOnlineFeaturesRequestV2( + features=_build_feature_references(feature_ref_strs=feature_refs), + entity_rows=_infer_online_entity_rows(entity_rows), + project=project if project is not None else self.project, + ), + metadata=self._get_grpc_metadata(), + ) + except grpc.RpcError as e: + raise grpc.RpcError(e.details()) + + response = OnlineResponse(response) + return response + + +def _build_feature_references(feature_ref_strs: List[str]) -> List[FeatureReferenceV2]: + """ + Builds a list of FeatureReference protos from a list of FeatureReference strings + + Args: + feature_ref_strs: List of string feature references + Returns: + A list of FeatureReference protos parsed from args. + """ + + feature_refs = [FeatureRef.from_str(ref_str) for ref_str in feature_ref_strs] + feature_ref_protos = [ref.to_proto() for ref in feature_refs] + + return feature_ref_protos + + +def _infer_online_entity_rows( + entity_rows: List[Dict[str, Any]] +) -> List[GetOnlineFeaturesRequestV2.EntityRow]: + """ + Builds a list of EntityRow protos from Python native type format passed by user. + + Args: + entity_rows: A list of dictionaries where each key-value is an entity-name, entity-value pair. + Returns: + A list of EntityRow protos parsed from args. + """ + + entity_rows_dicts = cast(List[Dict[str, Any]], entity_rows) + entity_row_list = [] + entity_type_map = dict() + + for entity in entity_rows_dicts: + fields = {} + for key, value in entity.items(): + # Allow for feast.types.Value + if isinstance(value, Value): + proto_value = value + else: + # Infer the specific type for this row + current_dtype = python_type_to_feast_value_type(name=key, value=value) + + if key not in entity_type_map: + entity_type_map[key] = current_dtype + else: + if current_dtype != entity_type_map[key]: + raise TypeError( + f"Input entity {key} has mixed types, {current_dtype} and {entity_type_map[key]}. That is not allowed. " + ) + proto_value = _python_value_to_proto_value(current_dtype, value) + fields[key] = proto_value + entity_row_list.append(GetOnlineFeaturesRequestV2.EntityRow(fields=fields)) + return entity_row_list diff --git a/sdk/python/feast/feature.py b/sdk/python/feast/feature.py index 7c6fdd74417..15c827bcf52 100644 --- a/sdk/python/feast/feature.py +++ b/sdk/python/feast/feature.py @@ -15,6 +15,7 @@ from typing import MutableMapping, Optional from feast.core.Feature_pb2 import FeatureSpecV2 as FeatureSpecProto +from feast.serving.ServingService_pb2 import FeatureReferenceV2 as FeatureRefProto from feast.types import Value_pb2 as ValueTypeProto from feast.value_type import ValueType @@ -95,3 +96,54 @@ def from_proto(cls, feature_proto: FeatureSpecProto): ) return feature + + +class FeatureRef: + """ Feature Reference represents a reference to a specific feature. """ + + def __init__(self, name: str, feature_table: str = None): + self.proto = FeatureRefProto(name=name, feature_table=feature_table) + + @classmethod + def from_proto(cls, proto: FeatureRefProto): + """ + Construct a feature reference from the given FeatureReference proto + Arg: + proto: Protobuf FeatureReference to construct from + Returns: + FeatureRef that refers to the given feature + """ + return cls(name=proto.name, feature_table=proto.feature_table) + + @classmethod + def from_str(cls, feature_ref_str: str): + """ + Parse the given string feature reference into FeatureRef model + String feature reference should be in the format feature_table:feature. + Where "feature_table" and "name" are the feature_table name and feature name + respectively. + Args: + feature_ref_str: String representation of the feature reference + Returns: + FeatureRef that refers to the given feature + """ + proto = FeatureRefProto() + + # parse feature table name if specified + if ":" in feature_ref_str: + proto.feature_table, proto.name = feature_ref_str.split(":") + else: + raise ValueError( + f"Unsupported feature reference: {feature_ref_str} - Feature reference string should be in the form [featuretable_name:featurename]" + ) + + return cls.from_proto(proto) + + def to_proto(self) -> FeatureRefProto: + """ + Convert and return this feature table reference to protobuf. + Returns: + Protobuf respresentation of this feature table reference. + """ + + return self.proto diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 3d7fbab7309..ad62ca823da 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -46,7 +46,12 @@ from feast.feature import Feature from feast.feature_table import FeatureTable from feast.serving import ServingService_pb2_grpc as Serving -from feast.serving.ServingService_pb2 import GetFeastServingInfoResponse +from feast.serving.ServingService_pb2 import FeatureReferenceV2 as FeatureRefProto +from feast.serving.ServingService_pb2 import ( + GetFeastServingInfoResponse, + GetOnlineFeaturesRequestV2, + GetOnlineFeaturesResponse, +) from feast.types import Value_pb2 as ValueProto from feast.value_type import ValueType from feast_core_server import ( @@ -321,24 +326,6 @@ def test_version(self, mocked_client, mocker): and status["serving"]["version"] == "0.3.2" ) - @pytest.mark.parametrize( - "mocked_client,auth_metadata", - [ - (lazy_fixture("mock_client"), ()), - (lazy_fixture("mock_client_with_auth"), (AUTH_METADATA)), - (lazy_fixture("secure_mock_client"), ()), - (lazy_fixture("secure_mock_client_with_auth"), (AUTH_METADATA)), - ], - ids=[ - "mock_client_without_auth", - "mock_client_with_auth", - "secure_mock_client_without_auth", - "secure_mock_client_with_auth", - ], - ) - def test_get_online_features(self, mocked_client, auth_metadata, mocker): - assert 1 == 1 - @pytest.mark.parametrize( "mocked_client", [ @@ -547,6 +534,99 @@ def test_ingest_csv(self, mocked_client, mocker, tmp_path): assert_frame_equal(partitioned_df, pq_df) + @pytest.mark.parametrize( + "mocked_client,auth_metadata", + [ + (lazy_fixture("mock_client"), ()), + (lazy_fixture("mock_client_with_auth"), (AUTH_METADATA)), + (lazy_fixture("secure_mock_client"), ()), + (lazy_fixture("secure_mock_client_with_auth"), (AUTH_METADATA)), + ], + ids=[ + "mock_client_without_auth", + "mock_client_with_auth", + "secure_mock_client_without_auth", + "secure_mock_client_with_auth", + ], + ) + def test_get_online_features(self, mocked_client, auth_metadata, mocker): + ROW_COUNT = 100 + + mocked_client._serving_service_stub = Serving.ServingServiceStub( + grpc.insecure_channel("") + ) + + def int_val(x): + return ValueProto.Value(int64_val=x) + + def string_val(x): + return ValueProto.Value(string_val=x) + + request = GetOnlineFeaturesRequestV2(project="driver_project") + request.features.extend( + [ + FeatureRefProto(feature_table="driver", name="age"), + FeatureRefProto(feature_table="driver", name="rating"), + FeatureRefProto(feature_table="driver", name="null_value"), + ] + ) + + receive_response = GetOnlineFeaturesResponse() + entity_rows = [] + for row_number in range(1, ROW_COUNT + 1): + request.entity_rows.append( + GetOnlineFeaturesRequestV2.EntityRow( + fields={"driver_id": int_val(row_number)} + ) + ) + entity_rows.append({"driver_id": int_val(row_number)}) + field_values = GetOnlineFeaturesResponse.FieldValues( + fields={ + "driver_id": int_val(row_number), + "driver:age": int_val(1), + "driver:rating": string_val("9"), + "driver:null_value": ValueProto.Value(), + }, + statuses={ + "driver_id": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:age": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:rating": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:null_value": GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE, + }, + ) + receive_response.field_values.append(field_values) + + mocker.patch.object( + mocked_client._serving_service_stub, + "GetOnlineFeaturesV2", + return_value=receive_response, + ) + got_response = mocked_client.get_online_features( + entity_rows=entity_rows, + feature_refs=["driver:age", "driver:rating", "driver:null_value"], + project="driver_project", + ) # type: GetOnlineFeaturesResponse + mocked_client._serving_service_stub.GetOnlineFeaturesV2.assert_called_with( + request, metadata=auth_metadata + ) + + got_fields = got_response.field_values[0].fields + got_statuses = got_response.field_values[0].statuses + assert ( + got_fields["driver_id"] == int_val(1) + and got_statuses["driver_id"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver:age"] == int_val(1) + and got_statuses["driver:age"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver:rating"] == string_val("9") + and got_statuses["driver:rating"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver:null_value"] == ValueProto.Value() + and got_statuses["driver:null_value"] + == GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE + ) + @patch("grpc.channel_ready_future") def test_secure_channel_creation_with_secure_client( self, _mocked_obj, core_server, serving_server From 04b04d4f9f9f04408888a7e2f7ac860973eaffd2 Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 14 Oct 2020 16:05:47 +0800 Subject: [PATCH 3/7] Update go protos Signed-off-by: Terence --- go.mod | 2 +- go.sum | 2 + sdk/go/protos/feast/core/DataSource.pb.go | 155 ++++++++++++---------- 3 files changed, 87 insertions(+), 72 deletions(-) diff --git a/go.mod b/go.mod index 6bb0e0a33cf..8b36dfe6788 100644 --- a/go.mod +++ b/go.mod @@ -25,7 +25,7 @@ require ( golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect golang.org/x/net v0.0.0-20200822124328-c89045814202 golang.org/x/sys v0.0.0-20200515095857-1151b9dac4a9 // indirect - golang.org/x/tools v0.0.0-20201011145850-ed2f50202694 // indirect + golang.org/x/tools v0.0.0-20201013201025-64a9e34f3752 // indirect google.golang.org/grpc v1.29.1 google.golang.org/protobuf v1.25.0 // indirect gopkg.in/russross/blackfriday.v2 v2.0.0 // indirect diff --git a/go.sum b/go.sum index 4fcb8cd56d0..4e26be46a3c 100644 --- a/go.sum +++ b/go.sum @@ -494,6 +494,8 @@ golang.org/x/tools v0.0.0-20201011145850-ed2f50202694 h1:BANdcOVw3KTuUiyfDp7wrzC golang.org/x/tools v0.0.0-20201011145850-ed2f50202694/go.mod h1:z6u4i615ZeAfBE4XtMziQW1fSVJXACjjbWkB/mvPzlU= golang.org/x/tools v0.0.0-20201013053347-2db1cd791039 h1:kLBxO4OPBgPwjg8Vvu+/0DCHIfDwYIGNFcD66NU9kpo= golang.org/x/tools v0.0.0-20201013053347-2db1cd791039/go.mod h1:z6u4i615ZeAfBE4XtMziQW1fSVJXACjjbWkB/mvPzlU= +golang.org/x/tools v0.0.0-20201013201025-64a9e34f3752 h1:2ntEwh02rqo2jSsrYmp4yKHHjh0CbXP3ZtSUetSB+q8= +golang.org/x/tools v0.0.0-20201013201025-64a9e34f3752/go.mod h1:z6u4i615ZeAfBE4XtMziQW1fSVJXACjjbWkB/mvPzlU= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= diff --git a/sdk/go/protos/feast/core/DataSource.pb.go b/sdk/go/protos/feast/core/DataSource.pb.go index 8a21f3c3551..a70176bccb9 100644 --- a/sdk/go/protos/feast/core/DataSource.pb.go +++ b/sdk/go/protos/feast/core/DataSource.pb.go @@ -107,11 +107,13 @@ type DataSource struct { // Defines mapping between fields in the sourced data // and fields in parent FeatureTable. FieldMapping map[string]string `protobuf:"bytes,2,rep,name=field_mapping,json=fieldMapping,proto3" json:"field_mapping,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` - // Must specify timestamp column name - TimestampColumn string `protobuf:"bytes,3,opt,name=timestamp_column,json=timestampColumn,proto3" json:"timestamp_column,omitempty"` + // Must specify event timestamp column name + EventTimestampColumn string `protobuf:"bytes,3,opt,name=event_timestamp_column,json=eventTimestampColumn,proto3" json:"event_timestamp_column,omitempty"` // (Optional) Specify partition column // useful for file sources DatePartitionColumn string `protobuf:"bytes,4,opt,name=date_partition_column,json=datePartitionColumn,proto3" json:"date_partition_column,omitempty"` + // Must specify creation timestamp column name + CreatedTimestampColumn string `protobuf:"bytes,5,opt,name=created_timestamp_column,json=createdTimestampColumn,proto3" json:"created_timestamp_column,omitempty"` // DataSource options. // // Types that are assignable to Options: @@ -168,9 +170,9 @@ func (x *DataSource) GetFieldMapping() map[string]string { return nil } -func (x *DataSource) GetTimestampColumn() string { +func (x *DataSource) GetEventTimestampColumn() string { if x != nil { - return x.TimestampColumn + return x.EventTimestampColumn } return "" } @@ -182,6 +184,13 @@ func (x *DataSource) GetDatePartitionColumn() string { return "" } +func (x *DataSource) GetCreatedTimestampColumn() string { + if x != nil { + return x.CreatedTimestampColumn + } + return "" +} + func (m *DataSource) GetOptions() isDataSource_Options { if m != nil { return m.Options @@ -500,7 +509,7 @@ var File_feast_core_DataSource_proto protoreflect.FileDescriptor var file_feast_core_DataSource_proto_rawDesc = []byte{ 0x0a, 0x1b, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2f, 0x63, 0x6f, 0x72, 0x65, 0x2f, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0a, 0x66, - 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x22, 0xb5, 0x08, 0x0a, 0x0a, 0x44, 0x61, + 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x22, 0xfa, 0x08, 0x0a, 0x0a, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x35, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x21, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x53, @@ -509,72 +518,76 @@ var file_feast_core_DataSource_proto_rawDesc = []byte{ 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x28, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x45, 0x6e, 0x74, 0x72, 0x79, - 0x52, 0x0c, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x12, 0x29, - 0x0a, 0x10, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x5f, 0x63, 0x6f, 0x6c, 0x75, - 0x6d, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, - 0x61, 0x6d, 0x70, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x12, 0x32, 0x0a, 0x15, 0x64, 0x61, 0x74, - 0x65, 0x5f, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x63, 0x6f, 0x6c, 0x75, - 0x6d, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x64, 0x61, 0x74, 0x65, 0x50, 0x61, - 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x12, 0x47, 0x0a, - 0x0c, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x0b, 0x20, - 0x01, 0x28, 0x0b, 0x32, 0x22, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, - 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x46, 0x69, 0x6c, 0x65, - 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0b, 0x66, 0x69, 0x6c, 0x65, 0x4f, - 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x53, 0x0a, 0x10, 0x62, 0x69, 0x67, 0x71, 0x75, 0x65, - 0x72, 0x79, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x0b, - 0x32, 0x26, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x44, 0x61, - 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x42, 0x69, 0x67, 0x51, 0x75, 0x65, 0x72, - 0x79, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0f, 0x62, 0x69, 0x67, 0x71, - 0x75, 0x65, 0x72, 0x79, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x4a, 0x0a, 0x0d, 0x6b, - 0x61, 0x66, 0x6b, 0x61, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x0d, 0x20, 0x01, - 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, - 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4b, 0x61, 0x66, 0x6b, 0x61, - 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0c, 0x6b, 0x61, 0x66, 0x6b, 0x61, - 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x50, 0x0a, 0x0f, 0x6b, 0x69, 0x6e, 0x65, 0x73, - 0x69, 0x73, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x0e, 0x20, 0x01, 0x28, 0x0b, - 0x32, 0x25, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x44, 0x61, - 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4b, 0x69, 0x6e, 0x65, 0x73, 0x69, 0x73, - 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0e, 0x6b, 0x69, 0x6e, 0x65, 0x73, - 0x69, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3f, 0x0a, 0x11, 0x46, 0x69, 0x65, - 0x6c, 0x64, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, - 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, - 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x1a, 0x49, 0x0a, 0x0b, 0x46, 0x69, - 0x6c, 0x65, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1f, 0x0a, 0x0b, 0x66, 0x69, 0x6c, - 0x65, 0x5f, 0x66, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, - 0x66, 0x69, 0x6c, 0x65, 0x46, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x12, 0x19, 0x0a, 0x08, 0x66, 0x69, - 0x6c, 0x65, 0x5f, 0x75, 0x72, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x66, 0x69, - 0x6c, 0x65, 0x55, 0x72, 0x6c, 0x1a, 0x2e, 0x0a, 0x0f, 0x42, 0x69, 0x67, 0x51, 0x75, 0x65, 0x72, - 0x79, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x74, 0x61, 0x62, 0x6c, - 0x65, 0x5f, 0x72, 0x65, 0x66, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x74, 0x61, 0x62, - 0x6c, 0x65, 0x52, 0x65, 0x66, 0x1a, 0x70, 0x0a, 0x0c, 0x4b, 0x61, 0x66, 0x6b, 0x61, 0x4f, 0x70, - 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x2b, 0x0a, 0x11, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, - 0x61, 0x70, 0x5f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, - 0x72, 0x73, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, - 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1d, 0x0a, 0x0a, 0x63, 0x6c, 0x61, 0x73, - 0x73, 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x63, 0x6c, - 0x61, 0x73, 0x73, 0x50, 0x61, 0x74, 0x68, 0x1a, 0x68, 0x0a, 0x0e, 0x4b, 0x69, 0x6e, 0x65, 0x73, - 0x69, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x16, 0x0a, 0x06, 0x72, 0x65, 0x67, - 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, - 0x6e, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x5f, 0x6e, 0x61, 0x6d, 0x65, - 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4e, 0x61, - 0x6d, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x63, 0x6c, 0x61, 0x73, 0x73, 0x5f, 0x70, 0x61, 0x74, 0x68, - 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x63, 0x6c, 0x61, 0x73, 0x73, 0x50, 0x61, 0x74, - 0x68, 0x22, 0x63, 0x0a, 0x0a, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, - 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x56, 0x41, 0x4c, 0x49, 0x44, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, - 0x42, 0x41, 0x54, 0x43, 0x48, 0x5f, 0x46, 0x49, 0x4c, 0x45, 0x10, 0x01, 0x12, 0x12, 0x0a, 0x0e, - 0x42, 0x41, 0x54, 0x43, 0x48, 0x5f, 0x42, 0x49, 0x47, 0x51, 0x55, 0x45, 0x52, 0x59, 0x10, 0x02, - 0x12, 0x10, 0x0a, 0x0c, 0x53, 0x54, 0x52, 0x45, 0x41, 0x4d, 0x5f, 0x4b, 0x41, 0x46, 0x4b, 0x41, - 0x10, 0x03, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, 0x52, 0x45, 0x41, 0x4d, 0x5f, 0x4b, 0x49, 0x4e, - 0x45, 0x53, 0x49, 0x53, 0x10, 0x04, 0x42, 0x09, 0x0a, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, - 0x73, 0x42, 0x58, 0x0a, 0x10, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, - 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x42, 0x0f, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, - 0x65, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, - 0x6f, 0x6d, 0x2f, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2d, 0x64, 0x65, 0x76, 0x2f, 0x66, 0x65, 0x61, - 0x73, 0x74, 0x2f, 0x73, 0x64, 0x6b, 0x2f, 0x67, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x73, - 0x2f, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2f, 0x63, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, - 0x74, 0x6f, 0x33, + 0x52, 0x0c, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x12, 0x34, + 0x0a, 0x16, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, + 0x70, 0x5f, 0x63, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x14, + 0x65, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x43, 0x6f, + 0x6c, 0x75, 0x6d, 0x6e, 0x12, 0x32, 0x0a, 0x15, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x70, 0x61, 0x72, + 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x63, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x18, 0x04, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x13, 0x64, 0x61, 0x74, 0x65, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, + 0x6f, 0x6e, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x12, 0x38, 0x0a, 0x18, 0x63, 0x72, 0x65, 0x61, + 0x74, 0x65, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x5f, 0x63, 0x6f, + 0x6c, 0x75, 0x6d, 0x6e, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x16, 0x63, 0x72, 0x65, 0x61, + 0x74, 0x65, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x43, 0x6f, 0x6c, 0x75, + 0x6d, 0x6e, 0x12, 0x47, 0x0a, 0x0c, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, + 0x6e, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x22, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, + 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, + 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0b, + 0x66, 0x69, 0x6c, 0x65, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x53, 0x0a, 0x10, 0x62, + 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, + 0x0c, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x26, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, + 0x72, 0x65, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x42, 0x69, + 0x67, 0x51, 0x75, 0x65, 0x72, 0x79, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, + 0x0f, 0x62, 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x12, 0x4a, 0x0a, 0x0d, 0x6b, 0x61, 0x66, 0x6b, 0x61, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, + 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, + 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, + 0x4b, 0x61, 0x66, 0x6b, 0x61, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0c, + 0x6b, 0x61, 0x66, 0x6b, 0x61, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x50, 0x0a, 0x0f, + 0x6b, 0x69, 0x6e, 0x65, 0x73, 0x69, 0x73, 0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, + 0x0e, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x25, 0x2e, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, 0x63, 0x6f, + 0x72, 0x65, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4b, 0x69, + 0x6e, 0x65, 0x73, 0x69, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x48, 0x00, 0x52, 0x0e, + 0x6b, 0x69, 0x6e, 0x65, 0x73, 0x69, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3f, + 0x0a, 0x11, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x4d, 0x61, 0x70, 0x70, 0x69, 0x6e, 0x67, 0x45, 0x6e, + 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x1a, + 0x49, 0x0a, 0x0b, 0x46, 0x69, 0x6c, 0x65, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1f, + 0x0a, 0x0b, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x66, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x0a, 0x66, 0x69, 0x6c, 0x65, 0x46, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x12, + 0x19, 0x0a, 0x08, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x75, 0x72, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x07, 0x66, 0x69, 0x6c, 0x65, 0x55, 0x72, 0x6c, 0x1a, 0x2e, 0x0a, 0x0f, 0x42, 0x69, + 0x67, 0x51, 0x75, 0x65, 0x72, 0x79, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1b, 0x0a, + 0x09, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x5f, 0x72, 0x65, 0x66, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x08, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x52, 0x65, 0x66, 0x1a, 0x70, 0x0a, 0x0c, 0x4b, 0x61, + 0x66, 0x6b, 0x61, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x2b, 0x0a, 0x11, 0x62, 0x6f, + 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x5f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, + 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1d, 0x0a, + 0x0a, 0x63, 0x6c, 0x61, 0x73, 0x73, 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18, 0x03, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x09, 0x63, 0x6c, 0x61, 0x73, 0x73, 0x50, 0x61, 0x74, 0x68, 0x1a, 0x68, 0x0a, 0x0e, + 0x4b, 0x69, 0x6e, 0x65, 0x73, 0x69, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x16, + 0x0a, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, + 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x74, 0x72, + 0x65, 0x61, 0x6d, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x63, 0x6c, 0x61, 0x73, 0x73, + 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x63, 0x6c, 0x61, + 0x73, 0x73, 0x50, 0x61, 0x74, 0x68, 0x22, 0x63, 0x0a, 0x0a, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, + 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x56, 0x41, 0x4c, 0x49, 0x44, 0x10, + 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x42, 0x41, 0x54, 0x43, 0x48, 0x5f, 0x46, 0x49, 0x4c, 0x45, 0x10, + 0x01, 0x12, 0x12, 0x0a, 0x0e, 0x42, 0x41, 0x54, 0x43, 0x48, 0x5f, 0x42, 0x49, 0x47, 0x51, 0x55, + 0x45, 0x52, 0x59, 0x10, 0x02, 0x12, 0x10, 0x0a, 0x0c, 0x53, 0x54, 0x52, 0x45, 0x41, 0x4d, 0x5f, + 0x4b, 0x41, 0x46, 0x4b, 0x41, 0x10, 0x03, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x54, 0x52, 0x45, 0x41, + 0x4d, 0x5f, 0x4b, 0x49, 0x4e, 0x45, 0x53, 0x49, 0x53, 0x10, 0x04, 0x42, 0x09, 0x0a, 0x07, 0x6f, + 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x42, 0x58, 0x0a, 0x10, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2e, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x42, 0x0f, 0x44, 0x61, 0x74, 0x61, + 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x33, 0x67, 0x69, 0x74, + 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2d, 0x64, 0x65, + 0x76, 0x2f, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2f, 0x73, 0x64, 0x6b, 0x2f, 0x67, 0x6f, 0x2f, 0x70, + 0x72, 0x6f, 0x74, 0x6f, 0x73, 0x2f, 0x66, 0x65, 0x61, 0x73, 0x74, 0x2f, 0x63, 0x6f, 0x72, 0x65, + 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( From 7ab4ca71ccad862f53dc70fc087d6ed4b07d234b Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 14 Oct 2020 16:11:52 +0800 Subject: [PATCH 4/7] Update go sdk Signed-off-by: Terence --- sdk/go/client.go | 2 +- sdk/go/client_test.go | 15 +++++------- sdk/go/request.go | 54 ++++++++++++++--------------------------- sdk/go/request_test.go | 29 +++++++++++++--------- sdk/go/response_test.go | 36 +++++++++++++-------------- 5 files changed, 60 insertions(+), 76 deletions(-) diff --git a/sdk/go/client.go b/sdk/go/client.go index 7accfcd0c8f..29e01d299b1 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -97,7 +97,7 @@ func (fc *GrpcClient) GetOnlineFeatures(ctx context.Context, req *OnlineFeatures if err != nil { return nil, err } - resp, err := fc.cli.GetOnlineFeatures(ctx, featuresRequest) + resp, err := fc.cli.GetOnlineFeaturesV2(ctx, featuresRequest) // collect unqiue entity refs from entity rows entityRefs := make(map[string]struct{}) diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index ab04cdeaa15..914592c44b6 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -26,8 +26,7 @@ func TestGetOnlineFeatures(t *testing.T) { req: OnlineFeaturesRequest{ Features: []string{ "driver:rating", - "rating", - "null_value", + "driver:null_value", }, Entities: []Row{ {"driver_id": Int64Val(1)}, @@ -39,14 +38,12 @@ func TestGetOnlineFeatures(t *testing.T) { FieldValues: []*serving.GetOnlineFeaturesResponse_FieldValues{ { Fields: map[string]*types.Value{ - "driver:rating": Int64Val(1), - "rating": Int64Val(1), - "null_value": {}, + "driver:rating": Int64Val(1), + "driver:null_value": {}, }, Statuses: map[string]serving.GetOnlineFeaturesResponse_FieldStatus{ - "driver:rating": serving.GetOnlineFeaturesResponse_PRESENT, - "rating": serving.GetOnlineFeaturesResponse_PRESENT, - "null_value": serving.GetOnlineFeaturesResponse_NULL_VALUE, + "driver:rating": serving.GetOnlineFeaturesResponse_PRESENT, + "driver:null_value": serving.GetOnlineFeaturesResponse_NULL_VALUE, }, }, }, @@ -65,7 +62,7 @@ func TestGetOnlineFeatures(t *testing.T) { _, traceCtx := opentracing.StartSpanFromContext(ctx, "get_online_features") rawRequest, _ := tc.req.buildRequest() resp := tc.want.RawResponse - cli.EXPECT().GetOnlineFeatures(traceCtx, rawRequest).Return(resp, nil).Times(1) + cli.EXPECT().GetOnlineFeaturesV2(traceCtx, rawRequest).Return(resp, nil).Times(1) client := &GrpcClient{ cli: cli, diff --git a/sdk/go/request.go b/sdk/go/request.go index 010545d6c4e..360603b3a3e 100644 --- a/sdk/go/request.go +++ b/sdk/go/request.go @@ -10,13 +10,13 @@ var ( // ErrInvalidFeatureRef indicates that the user has provided a feature reference // with the wrong structure or contents ErrInvalidFeatureRef = "Invalid Feature Reference %s provided, " + - "feature reference must be in the format [featureset:]name" + "feature reference must be in the format featureTableName:featureName" ) -// OnlineFeaturesRequest wrapper on feast.serving.GetOnlineFeaturesRequest. +// OnlineFeaturesRequest wrapper on feast.serving.GetOnlineFeaturesRequestV2. type OnlineFeaturesRequest struct { // Features is the list of features to obtain from Feast. Each feature can be given as - // the format feature_set:feature, where "feature_set" & "feature" are feature set name + // the format feature_table:feature, where "feature_table" & "feature" are feature table name // and feature name respectively. The only required components is feature name. Features []string @@ -26,41 +26,37 @@ type OnlineFeaturesRequest struct { // Project optionally specifies the project override. If specified, uses given project for retrieval. // Overrides the projects specified in Feature References if also specified. Project string - - // whether to omit the entities fields in the response. - OmitEntities bool } // Builds the feast-specified request payload from the wrapper. -func (r OnlineFeaturesRequest) buildRequest() (*serving.GetOnlineFeaturesRequest, error) { +func (r OnlineFeaturesRequest) buildRequest() (*serving.GetOnlineFeaturesRequestV2, error) { featureRefs, err := buildFeatureRefs(r.Features) if err != nil { return nil, err } // build request entity rows from native entities - entityRows := make([]*serving.GetOnlineFeaturesRequest_EntityRow, len(r.Entities)) + entityRows := make([]*serving.GetOnlineFeaturesRequestV2_EntityRow, len(r.Entities)) for i, entity := range r.Entities { - entityRows[i] = &serving.GetOnlineFeaturesRequest_EntityRow{ + entityRows[i] = &serving.GetOnlineFeaturesRequestV2_EntityRow{ Fields: entity, } } - return &serving.GetOnlineFeaturesRequest{ - Features: featureRefs, - EntityRows: entityRows, - OmitEntitiesInResponse: r.OmitEntities, - Project: r.Project, + return &serving.GetOnlineFeaturesRequestV2{ + Features: featureRefs, + EntityRows: entityRows, + Project: r.Project, }, nil } // Creates a slice of FeatureReferences from string representation in -// the format featureset:feature. +// the format featuretable:feature. // featureRefStrs - string feature references to parse. // Returns parsed FeatureReferences. // Returns an error when the format of the string feature reference is invalid -func buildFeatureRefs(featureRefStrs []string) ([]*serving.FeatureReference, error) { - var featureRefs []*serving.FeatureReference +func buildFeatureRefs(featureRefStrs []string) ([]*serving.FeatureReferenceV2, error) { + var featureRefs []*serving.FeatureReferenceV2 for _, featureRefStr := range featureRefStrs { featureRef, err := parseFeatureRef(featureRefStr) @@ -76,35 +72,21 @@ func buildFeatureRefs(featureRefStrs []string) ([]*serving.FeatureReference, err // featureRefStr - the string feature reference to parse. // Returns parsed FeatureReference. // Returns an error when the format of the string feature reference is invalid -func parseFeatureRef(featureRefStr string) (*serving.FeatureReference, error) { +func parseFeatureRef(featureRefStr string) (*serving.FeatureReferenceV2, error) { if len(featureRefStr) == 0 { return nil, fmt.Errorf(ErrInvalidFeatureRef, featureRefStr) } - var featureRef serving.FeatureReference - if strings.Contains(featureRefStr, "/") { + var featureRef serving.FeatureReferenceV2 + if strings.Contains(featureRefStr, "/") || !strings.Contains(featureRefStr, ":") { return nil, fmt.Errorf(ErrInvalidFeatureRef, featureRefStr) } - // parse featureset if specified + // parse featuretable if specified if strings.Contains(featureRefStr, ":") { refSplit := strings.Split(featureRefStr, ":") - featureRef.FeatureSet, featureRefStr = refSplit[0], refSplit[1] + featureRef.FeatureTable, featureRefStr = refSplit[0], refSplit[1] } featureRef.Name = featureRefStr return &featureRef, nil } - -// Converts a FeatureReference proto into a string -// featureRef - The FeatureReference to render as string -// Returns string representation of the given FeatureReference -func toFeatureRefStr(featureRef *serving.FeatureReference) string { - refStr := "" - // In protov3, unset string and default to "" - if len(featureRef.FeatureSet) > 0 { - refStr += featureRef.FeatureSet + ":" - } - refStr += featureRef.Name - - return refStr -} diff --git a/sdk/go/request_test.go b/sdk/go/request_test.go index 943fd37fee7..0e9b89d119a 100644 --- a/sdk/go/request_test.go +++ b/sdk/go/request_test.go @@ -13,7 +13,7 @@ func TestGetOnlineFeaturesRequest(t *testing.T) { tt := []struct { name string req OnlineFeaturesRequest - want *serving.GetOnlineFeaturesRequest + want *serving.GetOnlineFeaturesRequestV2 wantErr bool err error }{ @@ -22,7 +22,6 @@ func TestGetOnlineFeaturesRequest(t *testing.T) { req: OnlineFeaturesRequest{ Features: []string{ "driver:driver_id", - "driver_id", }, Entities: []Row{ {"entity1": Int64Val(1), "entity2": StrVal("bob")}, @@ -31,17 +30,14 @@ func TestGetOnlineFeaturesRequest(t *testing.T) { }, Project: "driver_project", }, - want: &serving.GetOnlineFeaturesRequest{ - Features: []*serving.FeatureReference{ + want: &serving.GetOnlineFeaturesRequestV2{ + Features: []*serving.FeatureReferenceV2{ { - FeatureSet: "driver", - Name: "driver_id", - }, - { - Name: "driver_id", + FeatureTable: "driver", + Name: "driver_id", }, }, - EntityRows: []*serving.GetOnlineFeaturesRequest_EntityRow{ + EntityRows: []*serving.GetOnlineFeaturesRequestV2_EntityRow{ { Fields: map[string]*types.Value{ "entity1": Int64Val(1), @@ -61,8 +57,7 @@ func TestGetOnlineFeaturesRequest(t *testing.T) { }, }, }, - OmitEntitiesInResponse: false, - Project: "driver_project", + Project: "driver_project", }, wantErr: false, err: nil, @@ -77,6 +72,16 @@ func TestGetOnlineFeaturesRequest(t *testing.T) { wantErr: true, err: fmt.Errorf(ErrInvalidFeatureRef, "/fs1:feature1"), }, + { + name: "invalid_feature_name", + req: OnlineFeaturesRequest{ + Features: []string{"feature1"}, + Entities: []Row{}, + Project: "my_project", + }, + wantErr: true, + err: fmt.Errorf(ErrInvalidFeatureRef, "feature1"), + }, } for _, tc := range tt { t.Run(tc.name, func(t *testing.T) { diff --git a/sdk/go/response_test.go b/sdk/go/response_test.go index 0949f24d679..a6176527451 100644 --- a/sdk/go/response_test.go +++ b/sdk/go/response_test.go @@ -13,22 +13,22 @@ var response = OnlineFeaturesResponse{ FieldValues: []*serving.GetOnlineFeaturesResponse_FieldValues{ { Fields: map[string]*types.Value{ - "project1/feature1": Int64Val(1), - "project1/feature2": {}, + "featuretable1:feature1": Int64Val(1), + "featuretable1:feature2": {}, }, Statuses: map[string]serving.GetOnlineFeaturesResponse_FieldStatus{ - "project1/feature1": serving.GetOnlineFeaturesResponse_PRESENT, - "project1/feature2": serving.GetOnlineFeaturesResponse_NULL_VALUE, + "featuretable1:feature1": serving.GetOnlineFeaturesResponse_PRESENT, + "featuretable1:feature2": serving.GetOnlineFeaturesResponse_NULL_VALUE, }, }, { Fields: map[string]*types.Value{ - "project1/feature1": Int64Val(2), - "project1/feature2": Int64Val(2), + "featuretable1:feature1": Int64Val(2), + "featuretable1:feature2": Int64Val(2), }, Statuses: map[string]serving.GetOnlineFeaturesResponse_FieldStatus{ - "project1/feature1": serving.GetOnlineFeaturesResponse_PRESENT, - "project1/feature2": serving.GetOnlineFeaturesResponse_PRESENT, + "featuretable1:feature1": serving.GetOnlineFeaturesResponse_PRESENT, + "featuretable1:feature2": serving.GetOnlineFeaturesResponse_PRESENT, }, }, }, @@ -38,8 +38,8 @@ var response = OnlineFeaturesResponse{ func TestOnlineFeaturesResponseToRow(t *testing.T) { actual := response.Rows() expected := []Row{ - {"project1/feature1": Int64Val(1), "project1/feature2": &types.Value{}}, - {"project1/feature1": Int64Val(2), "project1/feature2": Int64Val(2)}, + {"featuretable1:feature1": Int64Val(1), "featuretable1:feature2": &types.Value{}}, + {"featuretable1:feature1": Int64Val(2), "featuretable1:feature2": Int64Val(2)}, } if len(expected) != len(actual) { t.Errorf("expected: %v, got: %v", expected, actual) @@ -55,12 +55,12 @@ func TestOnlineFeaturesResponseoToStatuses(t *testing.T) { actual := response.Statuses() expected := []map[string]serving.GetOnlineFeaturesResponse_FieldStatus{ { - "project1/feature1": serving.GetOnlineFeaturesResponse_PRESENT, - "project1/feature2": serving.GetOnlineFeaturesResponse_NULL_VALUE, + "featuretable1:feature1": serving.GetOnlineFeaturesResponse_PRESENT, + "featuretable1:feature2": serving.GetOnlineFeaturesResponse_NULL_VALUE, }, { - "project1/feature1": serving.GetOnlineFeaturesResponse_PRESENT, - "project1/feature2": serving.GetOnlineFeaturesResponse_PRESENT, + "featuretable1:feature1": serving.GetOnlineFeaturesResponse_PRESENT, + "featuretable1:feature2": serving.GetOnlineFeaturesResponse_PRESENT, }, } if len(expected) != len(actual) { @@ -88,7 +88,7 @@ func TestOnlineFeaturesResponseToInt64Array(t *testing.T) { { name: "valid", args: args{ - order: []string{"project1/feature2", "project1/feature1"}, + order: []string{"featuretable1:feature2", "featuretable1:feature1"}, fillNa: []int64{-1, -1}, }, want: [][]int64{{-1, 1}, {2, 2}}, @@ -97,7 +97,7 @@ func TestOnlineFeaturesResponseToInt64Array(t *testing.T) { { name: "length mismatch", args: args{ - order: []string{"fs:feature2", "fs:feature1"}, + order: []string{"ft:feature2", "ft:feature1"}, fillNa: []int64{-1}, }, want: nil, @@ -107,12 +107,12 @@ func TestOnlineFeaturesResponseToInt64Array(t *testing.T) { { name: "length mismatch", args: args{ - order: []string{"project1/feature2", "project1/feature3"}, + order: []string{"featuretable1:feature2", "featuretable1:feature3"}, fillNa: []int64{-1, -1}, }, want: nil, wantErr: true, - err: fmt.Errorf(ErrFeatureNotFound, "project1/feature3"), + err: fmt.Errorf(ErrFeatureNotFound, "featuretable1:feature3"), }, } for _, tc := range tt { From 53b0ffcb006c792c0b7c461916949f8c69b7023f Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 15 Oct 2020 10:47:52 +0800 Subject: [PATCH 5/7] Allow project to be unspecified for default proj Signed-off-by: Terence --- .../main/java/com/gojek/feast/FeastClient.java | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/sdk/java/src/main/java/com/gojek/feast/FeastClient.java b/sdk/java/src/main/java/com/gojek/feast/FeastClient.java index f8414b4429c..a5ea279e822 100644 --- a/sdk/java/src/main/java/com/gojek/feast/FeastClient.java +++ b/sdk/java/src/main/java/com/gojek/feast/FeastClient.java @@ -108,6 +108,21 @@ public GetFeastServingInfoResponse getFeastServingInfo() { return stub.getFeastServingInfo(GetFeastServingInfoRequest.newBuilder().build()); } + /** + * Get online features from Feast, without indicating project, will use `default`. + * + *

See {@link #getOnlineFeatures(List, List, String)} + * + * @param featureRefs list of string feature references to retrieve in the following format + * featureTable:feature, where 'featureTable' and 'feature' refer to the FeatureTable and + * Feature names respectively. Only the Feature name is required. + * @param rows list of {@link Row} to select the entities to retrieve the features for. + * @return list of {@link Row} containing retrieved data fields. + */ + public List getOnlineFeatures(List featureRefs, List rows) { + return getOnlineFeatures(featureRefs, rows, ""); + } + /** * Get online features from Feast. * @@ -127,7 +142,7 @@ public GetFeastServingInfoResponse getFeastServingInfo() { * featureTable:feature, where 'featureTable' and 'feature' refer to the FeatureTable and * Feature names respectively. Only the Feature name is required. * @param rows list of {@link Row} to select the entities to retrieve the features for - * @param project {@link String} Specifies the project override. If specifed uses the project for + * @param project {@link String} Specifies the project override. If specified uses the project for * retrieval. Overrides the projects set in Feature References if also specified. * @return list of {@link Row} containing retrieved data fields. */ From bd88ec011a0ae2d37fca01aac6caeb0682702d72 Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 15 Oct 2020 12:53:16 +0800 Subject: [PATCH 6/7] Address PR comments Signed-off-by: Terence --- sdk/python/feast/client.py | 19 +----- sdk/python/feast/feature.py | 18 +++++- sdk/python/tests/test_client.py | 105 ++++++++++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 19 deletions(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index d012396598f..684a2a68d00 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -56,7 +56,7 @@ from feast.core.CoreService_pb2_grpc import CoreServiceStub from feast.data_source import BigQuerySource, FileSource from feast.entity import Entity -from feast.feature import FeatureRef +from feast.feature import _build_feature_references from feast.feature_table import FeatureTable from feast.grpc import auth as feast_auth from feast.grpc.grpc import create_grpc_channel @@ -71,7 +71,6 @@ ) from feast.online_response import OnlineResponse from feast.serving.ServingService_pb2 import ( - FeatureReferenceV2, GetFeastServingInfoRequest, GetOnlineFeaturesRequestV2, ) @@ -772,22 +771,6 @@ def get_online_features( return response -def _build_feature_references(feature_ref_strs: List[str]) -> List[FeatureReferenceV2]: - """ - Builds a list of FeatureReference protos from a list of FeatureReference strings - - Args: - feature_ref_strs: List of string feature references - Returns: - A list of FeatureReference protos parsed from args. - """ - - feature_refs = [FeatureRef.from_str(ref_str) for ref_str in feature_ref_strs] - feature_ref_protos = [ref.to_proto() for ref in feature_refs] - - return feature_ref_protos - - def _infer_online_entity_rows( entity_rows: List[Dict[str, Any]] ) -> List[GetOnlineFeaturesRequestV2.EntityRow]: diff --git a/sdk/python/feast/feature.py b/sdk/python/feast/feature.py index 15c827bcf52..1d0e525a89b 100644 --- a/sdk/python/feast/feature.py +++ b/sdk/python/feast/feature.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import MutableMapping, Optional +from typing import List, MutableMapping, Optional from feast.core.Feature_pb2 import FeatureSpecV2 as FeatureSpecProto from feast.serving.ServingService_pb2 import FeatureReferenceV2 as FeatureRefProto @@ -147,3 +147,19 @@ def to_proto(self) -> FeatureRefProto: """ return self.proto + + +def _build_feature_references(feature_ref_strs: List[str]) -> List[FeatureRefProto]: + """ + Builds a list of FeatureReference protos from a list of FeatureReference strings + + Args: + feature_ref_strs: List of string feature references + Returns: + A list of FeatureReference protos parsed from args. + """ + + feature_refs = [FeatureRef.from_str(ref_str) for ref_str in feature_ref_strs] + feature_ref_protos = [ref.to_proto() for ref in feature_refs] + + return feature_ref_protos diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index ad62ca823da..de57cab6396 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -627,6 +627,111 @@ def string_val(x): == GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE ) + @pytest.mark.parametrize( + "mocked_client,auth_metadata", + [ + (lazy_fixture("mock_client"), ()), + (lazy_fixture("mock_client_with_auth"), (AUTH_METADATA)), + (lazy_fixture("secure_mock_client"), ()), + (lazy_fixture("secure_mock_client_with_auth"), (AUTH_METADATA)), + ], + ids=[ + "mock_client_without_auth", + "mock_client_with_auth", + "secure_mock_client_without_auth", + "secure_mock_client_with_auth", + ], + ) + def test_get_online_features_multi_entities( + self, mocked_client, auth_metadata, mocker + ): + ROW_COUNT = 100 + + mocked_client._serving_service_stub = Serving.ServingServiceStub( + grpc.insecure_channel("") + ) + + def int_val(x): + return ValueProto.Value(int64_val=x) + + def string_val(x): + return ValueProto.Value(string_val=x) + + request = GetOnlineFeaturesRequestV2(project="driver_project") + request.features.extend( + [ + FeatureRefProto(feature_table="driver", name="age"), + FeatureRefProto(feature_table="driver", name="rating"), + FeatureRefProto(feature_table="driver", name="null_value"), + ] + ) + + receive_response = GetOnlineFeaturesResponse() + entity_rows = [] + for row_number in range(1, ROW_COUNT + 1): + request.entity_rows.append( + GetOnlineFeaturesRequestV2.EntityRow( + fields={ + "driver_id": int_val(row_number), + "driver_id2": int_val(row_number), + } + ) + ) + entity_rows.append( + {"driver_id": int_val(row_number), "driver_id2": int_val(row_number)} + ) + field_values = GetOnlineFeaturesResponse.FieldValues( + fields={ + "driver_id": int_val(row_number), + "driver_id2": int_val(row_number), + "driver:age": int_val(1), + "driver:rating": string_val("9"), + "driver:null_value": ValueProto.Value(), + }, + statuses={ + "driver_id": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver_id2": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:age": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:rating": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:null_value": GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE, + }, + ) + receive_response.field_values.append(field_values) + + mocker.patch.object( + mocked_client._serving_service_stub, + "GetOnlineFeaturesV2", + return_value=receive_response, + ) + got_response = mocked_client.get_online_features( + entity_rows=entity_rows, + feature_refs=["driver:age", "driver:rating", "driver:null_value"], + project="driver_project", + ) # type: GetOnlineFeaturesResponse + mocked_client._serving_service_stub.GetOnlineFeaturesV2.assert_called_with( + request, metadata=auth_metadata + ) + + got_fields = got_response.field_values[0].fields + got_statuses = got_response.field_values[0].statuses + assert ( + got_fields["driver_id"] == int_val(1) + and got_statuses["driver_id"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver_id2"] == int_val(1) + and got_statuses["driver_id2"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver:age"] == int_val(1) + and got_statuses["driver:age"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver:rating"] == string_val("9") + and got_statuses["driver:rating"] + == GetOnlineFeaturesResponse.FieldStatus.PRESENT + and got_fields["driver:null_value"] == ValueProto.Value() + and got_statuses["driver:null_value"] + == GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE + ) + @patch("grpc.channel_ready_future") def test_secure_channel_creation_with_secure_client( self, _mocked_obj, core_server, serving_server From 2c266af1f6cfd2db83988bffe908f42a47f82ca0 Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 15 Oct 2020 14:44:09 +0800 Subject: [PATCH 7/7] Some cleanups Signed-off-by: Terence --- sdk/python/feast/client.py | 45 +---------- sdk/python/feast/online_response.py | 53 +++++++++++- sdk/python/tests/test_client.py | 121 ++++++++++++++-------------- 3 files changed, 111 insertions(+), 108 deletions(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index 684a2a68d00..22e5dbc28e9 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -14,7 +14,7 @@ import logging import multiprocessing import shutil -from typing import Any, Dict, List, Optional, Union, cast +from typing import Any, Dict, List, Optional, Union import grpc import pandas as pd @@ -69,14 +69,12 @@ _write_non_partitioned_table_from_source, _write_partitioned_table_from_source, ) -from feast.online_response import OnlineResponse +from feast.online_response import OnlineResponse, _infer_online_entity_rows from feast.serving.ServingService_pb2 import ( GetFeastServingInfoRequest, GetOnlineFeaturesRequestV2, ) from feast.serving.ServingService_pb2_grpc import ServingServiceStub -from feast.type_map import _python_value_to_proto_value, python_type_to_feast_value_type -from feast.types.Value_pb2 import Value as Value _logger = logging.getLogger(__name__) @@ -769,42 +767,3 @@ def get_online_features( response = OnlineResponse(response) return response - - -def _infer_online_entity_rows( - entity_rows: List[Dict[str, Any]] -) -> List[GetOnlineFeaturesRequestV2.EntityRow]: - """ - Builds a list of EntityRow protos from Python native type format passed by user. - - Args: - entity_rows: A list of dictionaries where each key-value is an entity-name, entity-value pair. - Returns: - A list of EntityRow protos parsed from args. - """ - - entity_rows_dicts = cast(List[Dict[str, Any]], entity_rows) - entity_row_list = [] - entity_type_map = dict() - - for entity in entity_rows_dicts: - fields = {} - for key, value in entity.items(): - # Allow for feast.types.Value - if isinstance(value, Value): - proto_value = value - else: - # Infer the specific type for this row - current_dtype = python_type_to_feast_value_type(name=key, value=value) - - if key not in entity_type_map: - entity_type_map[key] = current_dtype - else: - if current_dtype != entity_type_map[key]: - raise TypeError( - f"Input entity {key} has mixed types, {current_dtype} and {entity_type_map[key]}. That is not allowed. " - ) - proto_value = _python_value_to_proto_value(current_dtype, value) - fields[key] = proto_value - entity_row_list.append(GetOnlineFeaturesRequestV2.EntityRow(fields=fields)) - return entity_row_list diff --git a/sdk/python/feast/online_response.py b/sdk/python/feast/online_response.py index a90c25e09dd..6a86d9a63a2 100644 --- a/sdk/python/feast/online_response.py +++ b/sdk/python/feast/online_response.py @@ -12,10 +12,18 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Any, Dict, List +from typing import Any, Dict, List, cast -from feast.serving.ServingService_pb2 import GetOnlineFeaturesResponse -from feast.type_map import feast_value_type_to_python_type +from feast.serving.ServingService_pb2 import ( + GetOnlineFeaturesRequestV2, + GetOnlineFeaturesResponse, +) +from feast.type_map import ( + _python_value_to_proto_value, + feast_value_type_to_python_type, + python_type_to_feast_value_type, +) +from feast.types.Value_pb2 import Value as Value class OnlineResponse: @@ -52,3 +60,42 @@ def to_dict(self) -> Dict[str, Any]: features_dict[feature].append(native_type_value) return features_dict + + +def _infer_online_entity_rows( + entity_rows: List[Dict[str, Any]] +) -> List[GetOnlineFeaturesRequestV2.EntityRow]: + """ + Builds a list of EntityRow protos from Python native type format passed by user. + + Args: + entity_rows: A list of dictionaries where each key-value is an entity-name, entity-value pair. + Returns: + A list of EntityRow protos parsed from args. + """ + + entity_rows_dicts = cast(List[Dict[str, Any]], entity_rows) + entity_row_list = [] + entity_type_map = dict() + + for entity in entity_rows_dicts: + fields = {} + for key, value in entity.items(): + # Allow for feast.types.Value + if isinstance(value, Value): + proto_value = value + else: + # Infer the specific type for this row + current_dtype = python_type_to_feast_value_type(name=key, value=value) + + if key not in entity_type_map: + entity_type_map[key] = current_dtype + else: + if current_dtype != entity_type_map[key]: + raise TypeError( + f"Input entity {key} has mixed types, {current_dtype} and {entity_type_map[key]}. That is not allowed. " + ) + proto_value = _python_value_to_proto_value(current_dtype, value) + fields[key] = proto_value + entity_row_list.append(GetOnlineFeaturesRequestV2.EntityRow(fields=fields)) + return entity_row_list diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index de57cab6396..1d970907d50 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -294,6 +294,29 @@ def non_partitioned_df(self): } ) + @pytest.fixture + def get_online_features_fields_statuses(self): + ROW_COUNT = 100 + fields_statuses_tuple_list = [] + for row_number in range(0, ROW_COUNT): + fields_statuses_tuple_list.append( + ( + { + "driver_id": ValueProto.Value(int64_val=row_number), + "driver:age": ValueProto.Value(int64_val=1), + "driver:rating": ValueProto.Value(string_val="9"), + "driver:null_value": ValueProto.Value(), + }, + { + "driver_id": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:age": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:rating": GetOnlineFeaturesResponse.FieldStatus.PRESENT, + "driver:null_value": GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE, + }, + ) + ) + return fields_statuses_tuple_list + @pytest.mark.parametrize( "mocked_client", [lazy_fixture("mock_client"), lazy_fixture("secure_mock_client")], @@ -549,19 +572,15 @@ def test_ingest_csv(self, mocked_client, mocker, tmp_path): "secure_mock_client_with_auth", ], ) - def test_get_online_features(self, mocked_client, auth_metadata, mocker): + def test_get_online_features( + self, mocked_client, auth_metadata, mocker, get_online_features_fields_statuses + ): ROW_COUNT = 100 mocked_client._serving_service_stub = Serving.ServingServiceStub( grpc.insecure_channel("") ) - def int_val(x): - return ValueProto.Value(int64_val=x) - - def string_val(x): - return ValueProto.Value(string_val=x) - request = GetOnlineFeaturesRequestV2(project="driver_project") request.features.extend( [ @@ -573,28 +592,18 @@ def string_val(x): receive_response = GetOnlineFeaturesResponse() entity_rows = [] - for row_number in range(1, ROW_COUNT + 1): + for row_number in range(0, ROW_COUNT): + fields = get_online_features_fields_statuses[row_number][0] + statuses = get_online_features_fields_statuses[row_number][1] request.entity_rows.append( GetOnlineFeaturesRequestV2.EntityRow( - fields={"driver_id": int_val(row_number)} + fields={"driver_id": ValueProto.Value(int64_val=row_number)} ) ) - entity_rows.append({"driver_id": int_val(row_number)}) - field_values = GetOnlineFeaturesResponse.FieldValues( - fields={ - "driver_id": int_val(row_number), - "driver:age": int_val(1), - "driver:rating": string_val("9"), - "driver:null_value": ValueProto.Value(), - }, - statuses={ - "driver_id": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver:age": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver:rating": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver:null_value": GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE, - }, + entity_rows.append({"driver_id": ValueProto.Value(int64_val=row_number)}) + receive_response.field_values.append( + GetOnlineFeaturesResponse.FieldValues(fields=fields, statuses=statuses) ) - receive_response.field_values.append(field_values) mocker.patch.object( mocked_client._serving_service_stub, @@ -610,16 +619,16 @@ def string_val(x): request, metadata=auth_metadata ) - got_fields = got_response.field_values[0].fields - got_statuses = got_response.field_values[0].statuses + got_fields = got_response.field_values[1].fields + got_statuses = got_response.field_values[1].statuses assert ( - got_fields["driver_id"] == int_val(1) + got_fields["driver_id"] == ValueProto.Value(int64_val=1) and got_statuses["driver_id"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT - and got_fields["driver:age"] == int_val(1) + and got_fields["driver:age"] == ValueProto.Value(int64_val=1) and got_statuses["driver:age"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT - and got_fields["driver:rating"] == string_val("9") + and got_fields["driver:rating"] == ValueProto.Value(string_val="9") and got_statuses["driver:rating"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT and got_fields["driver:null_value"] == ValueProto.Value() @@ -643,7 +652,7 @@ def string_val(x): ], ) def test_get_online_features_multi_entities( - self, mocked_client, auth_metadata, mocker + self, mocked_client, auth_metadata, mocker, get_online_features_fields_statuses ): ROW_COUNT = 100 @@ -651,12 +660,6 @@ def test_get_online_features_multi_entities( grpc.insecure_channel("") ) - def int_val(x): - return ValueProto.Value(int64_val=x) - - def string_val(x): - return ValueProto.Value(string_val=x) - request = GetOnlineFeaturesRequestV2(project="driver_project") request.features.extend( [ @@ -668,35 +671,29 @@ def string_val(x): receive_response = GetOnlineFeaturesResponse() entity_rows = [] - for row_number in range(1, ROW_COUNT + 1): + for row_number in range(0, ROW_COUNT): + fields = get_online_features_fields_statuses[row_number][0] + fields["driver_id2"] = ValueProto.Value(int64_val=1) + statuses = get_online_features_fields_statuses[row_number][1] + statuses["driver_id2"] = GetOnlineFeaturesResponse.FieldStatus.PRESENT + request.entity_rows.append( GetOnlineFeaturesRequestV2.EntityRow( fields={ - "driver_id": int_val(row_number), - "driver_id2": int_val(row_number), + "driver_id": ValueProto.Value(int64_val=row_number), + "driver_id2": ValueProto.Value(int64_val=row_number), } ) ) entity_rows.append( - {"driver_id": int_val(row_number), "driver_id2": int_val(row_number)} + { + "driver_id": ValueProto.Value(int64_val=row_number), + "driver_id2": ValueProto.Value(int64_val=row_number), + } ) - field_values = GetOnlineFeaturesResponse.FieldValues( - fields={ - "driver_id": int_val(row_number), - "driver_id2": int_val(row_number), - "driver:age": int_val(1), - "driver:rating": string_val("9"), - "driver:null_value": ValueProto.Value(), - }, - statuses={ - "driver_id": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver_id2": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver:age": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver:rating": GetOnlineFeaturesResponse.FieldStatus.PRESENT, - "driver:null_value": GetOnlineFeaturesResponse.FieldStatus.NULL_VALUE, - }, + receive_response.field_values.append( + GetOnlineFeaturesResponse.FieldValues(fields=fields, statuses=statuses) ) - receive_response.field_values.append(field_values) mocker.patch.object( mocked_client._serving_service_stub, @@ -712,19 +709,19 @@ def string_val(x): request, metadata=auth_metadata ) - got_fields = got_response.field_values[0].fields - got_statuses = got_response.field_values[0].statuses + got_fields = got_response.field_values[1].fields + got_statuses = got_response.field_values[1].statuses assert ( - got_fields["driver_id"] == int_val(1) + got_fields["driver_id"] == ValueProto.Value(int64_val=1) and got_statuses["driver_id"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT - and got_fields["driver_id2"] == int_val(1) + and got_fields["driver_id2"] == ValueProto.Value(int64_val=1) and got_statuses["driver_id2"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT - and got_fields["driver:age"] == int_val(1) + and got_fields["driver:age"] == ValueProto.Value(int64_val=1) and got_statuses["driver:age"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT - and got_fields["driver:rating"] == string_val("9") + and got_fields["driver:rating"] == ValueProto.Value(string_val="9") and got_statuses["driver:rating"] == GetOnlineFeaturesResponse.FieldStatus.PRESENT and got_fields["driver:null_value"] == ValueProto.Value()