FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

perf: Replace MessageToDict with optimized custom dict builder (#6015) · agent-local-cloud/feast@9902064 · GitHub

Repository navigation

Commit 9902064

Browse files
perf: Replace MessageToDict with optimized custom dict builder (feast-dev#6015)
1 parent dda43c2 commit 9902064

3 files changed

Lines changed: 764 additions & 11 deletions

File tree

‎sdk/python/feast/feature_server.py‎

Lines changed: 5 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,6 @@
4040
from fastapi.logger import logger
4141
from fastapi.responses import JSONResponse
4242
from fastapi.staticfiles import StaticFiles
43-
from google.protobuf.json_format import MessageToDict
4443
from pydantic import BaseModel, field_validator
4544

4645
import feast
@@ -52,6 +51,7 @@
5251
FeastError,
5352
)
5453
from feast.feast_object import FeastObject
54+
from feast.feature_server_utils import convert_response_to_dict
5555
from feast.feature_view_utils import get_feature_view_from_feature_store
5656
from feast.permissions.action import WRITE, AuthzedAction
5757
from feast.permissions.security_manager import assert_permissions
@@ -392,12 +392,9 @@ async def get_online_features(request: GetOnlineFeaturesRequest) -> Any:
392392
)
393393

394394
response_dict = await run_in_threadpool(
395-
MessageToDict,
396-
response.proto,
397-
preserving_proto_field_name=True,
398-
float_precision=18,
395+
convert_response_to_dict, response.proto
399396
)
400-
return response_dict
397+
return JSONResponse(content=response_dict)
401398

402399
@app.post(
403400
"/retrieve-online-documents",
@@ -434,12 +431,9 @@ async def retrieve_online_documents(
434431
)
435432

436433
response_dict = await run_in_threadpool(
437-
MessageToDict,
438-
response.proto,
439-
preserving_proto_field_name=True,
440-
float_precision=18,
434+
convert_response_to_dict, response.proto
441435
)
442-
return response_dict
436+
return JSONResponse(content=response_dict)
443437

444438
@app.post("/push", dependencies=[Depends(inject_user_details)])
445439
async def push(request: PushFeaturesRequest) -> Response:
Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
"""Fast serialization utilities for Feature Server responses.
2+
3+
Matches the output format of MessageToDict with proto_json.patch() applied.
4+
Values are serialized as native Python types (not wrapped dicts).
5+
"""
6+
7+
import base64
8+
import logging
9+
from datetime import datetime, timezone
10+
from typing import Any, Dict, Optional
11+
12+
from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesResponse
13+
from feast.protos.feast.types.Value_pb2 import Value
14+
15+
logger = logging.getLogger(__name__)
16+
17+
# FieldStatus enum mapping (protos/feast/serving/ServingService.proto)
18+
_STATUS_NAMES: Dict[int, str] = {
19+
0: "INVALID",
20+
1: "PRESENT",
21+
2: "NULL_VALUE",
22+
3: "NOT_FOUND",
23+
4: "OUTSIDE_MAX_AGE",
24+
}
25+
26+
27+
def convert_response_to_dict(response: GetOnlineFeaturesResponse) -> Dict[str, Any]:
28+
"""Convert GetOnlineFeaturesResponse to a JSON-serializable dict.
29+
30+
Matches the structure produced by MessageToDict(proto, preserving_proto_field_name=True)
31+
with proto_json.patch() applied, with one intentional difference:
32+
33+
- double_val fields are returned as Python float objects (json.dumps uses Python 3.1+
34+
shortest round-trip form, ~15-17 sig digits) rather than 18 fixed significant digits
35+
(float_precision=18). Values are numerically identical; only the JSON string length
36+
may differ. This is safe for all ML feature types and avoids unnecessary precision
37+
overhead.
38+
"""
39+
result: Dict[str, Any] = {
40+
"results": [
41+
{
42+
"values": [_value_to_native(v) for v in feature_vector.values],
43+
"statuses": [
44+
_STATUS_NAMES.get(s, "INVALID") for s in feature_vector.statuses
45+
],
46+
**(
47+
{
48+
"event_timestamps": [
49+
_timestamp_to_str(ts)
50+
for ts in feature_vector.event_timestamps
51+
]
52+
}
53+
if feature_vector.event_timestamps
54+
else {}
55+
),
56+
}
57+
for feature_vector in response.results
58+
]
59+
}
60+
61+
if response.HasField("metadata"):
62+
result["metadata"] = _metadata_to_dict(response.metadata)
63+
64+
if response.status:
65+
result["status"] = response.status
66+
67+
return result
68+
69+
70+
def _value_to_native(v: Value) -> Optional[Any]:
71+
"""Convert a Value proto to a JSON-serializable Python type.
72+
73+
bytes_val and bytes_list_val are base64-encoded (RFC 4648) so that
74+
JSONResponse can serialize them without TypeError. This matches standard
75+
protobuf JSON encoding for bytes fields and is safe for all HTTP clients.
76+
"""
77+
which = v.WhichOneof("val")
78+
if which is None or which == "null_val":
79+
return None
80+
# bytes must be base64-encoded for JSON serialization
81+
elif which == "bytes_val":
82+
return base64.b64encode(v.bytes_val).decode("ascii")
83+
# RepeatedValue — nested Values that must be recursively converted
84+
elif which in ("list_val", "set_val"):
85+
return [_value_to_native(nested) for nested in getattr(v, which).val]
86+
# Map<string, Value> — recursively convert nested Values
87+
elif which in ("map_val", "struct_val"):
88+
return {k: _value_to_native(vv) for k, vv in getattr(v, which).val.items()}
89+
# MapList — list of Map<string, Value>
90+
elif which in ("map_list_val", "struct_list_val"):
91+
return [
92+
{k: _value_to_native(vv) for k, vv in m.val.items()}
93+
for m in getattr(v, which).val
94+
]
95+
# scalar_map_val has non-string keys; full conversion requires extra work and
96+
# this type is not returned by standard get_online_features paths today.
97+
elif which == "scalar_map_val":
98+
logger.warning(
99+
"scalar_map_val is not yet supported by convert_response_to_dict; value will be None"
100+
)
101+
return None
102+
# bytes_list_val / bytes_set_val — base64-encode each element
103+
elif which in ("bytes_list_val", "bytes_set_val"):
104+
return [base64.b64encode(b).decode("ascii") for b in getattr(v, which).val]
105+
# All other list/set types have scalar .val fields
106+
elif "_list_" in which or "_set_" in which:
107+
return list(getattr(v, which).val)
108+
else:
109+
return getattr(v, which)
110+
111+
112+
def _timestamp_to_str(ts) -> str:
113+
"""Convert protobuf Timestamp to RFC 3339 format with Z suffix.
114+
115+
Uses adaptive precision to match MessageToDict output:
116+
- No fractional seconds when nanos == 0
117+
- 3 digits (milliseconds) when nanos % 1_000_000 == 0
118+
- 6 digits (microseconds) when nanos % 1_000 == 0
119+
- 9 digits (nanoseconds) otherwise
120+
"""
121+
if ts.seconds == 0 and ts.nanos == 0:
122+
return "1970-01-01T00:00:00Z"
123+
dt = datetime.fromtimestamp(ts.seconds, tz=timezone.utc)
124+
base = dt.strftime("%Y-%m-%dT%H:%M:%S")
125+
nanos = ts.nanos
126+
if nanos == 0:
127+
return base + "Z"
128+
elif nanos % 1_000_000 == 0:
129+
return base + ".%03dZ" % (nanos // 1_000_000)
130+
elif nanos % 1_000 == 0:
131+
return base + ".%06dZ" % (nanos // 1_000)
132+
else:
133+
return base + ".%09dZ" % nanos
134+
135+
136+
def _metadata_to_dict(metadata) -> Dict[str, Any]:
137+
"""Convert FeatureResponseMeta to dict (matches proto_json.patch() format)."""
138+
result: Dict[str, Any] = {}
139+
if metadata.HasField("feature_names"):
140+
result["feature_names"] = list(metadata.feature_names.val)
141+
return result

0 commit comments

Comments
 (0)

Back | FazBrowse Home | New Git URL