Skip to content

Commit 6e9a532

Browse files
authored
feat: Add optional GLIDE client for Redis online reads (#6870)
* feat: Add optional GLIDE client for Redis online reads Adds a `client: glide` option to RedisOnlineStoreConfig. When set, the synchronous read paths (online_read and the batched read behind get_online_features) issue their HMGET commands as one non-atomic GLIDE batch, so the fetch runs off the GIL instead of in Python per command. Default behavior is unchanged: redis-py remains the default client, all writes and async reads keep using redis-py, and GLIDE reads the same entity keys and hashed feature fields Feast already writes. valkey-glide-sync is an optional dependency behind the new `feast[glide]` extra. Configuring `client: glide` without it installed raises a Feast extras import error naming the extra to install. This does not include the HGETALL threshold change from the same issue. Related to #6856. Signed-off-by: Chandler King <[email protected]> * fix: Release the GLIDE client and cache its module per store Addresses review feedback on the GLIDE read path: - teardown() now closes the native GLIDE client and clears the cached reference, so its Rust-side resources and file descriptors are released. - The glide_sync module is imported once per store and cached on the instance instead of being looked up on every read. - Allowlist the test password literals so detect-secrets stops failing lint-python. Signed-off-by: Chandler King <[email protected]> --------- Signed-off-by: Chandler King <[email protected]>
1 parent 81673d9 commit 6e9a532

5 files changed

Lines changed: 800 additions & 23 deletions

File tree

‎docs/reference/online-stores/redis.md‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,34 @@ Benchmark results against Redis 8.6.2 (localhost, 50 entities, 3 features/FV, 30
112112

113113
The speedup grows with the number of feature views and is most pronounced in production environments with non-trivial network RTT to Redis.
114114

115+
### GLIDE client (`client: glide`)
116+
117+
By default the Redis online store uses redis-py, which packs every command and parses every reply in Python. Each socket receive releases and reacquires the GIL, so in a process running other Python threads each reacquire waits on the switch interval. With many entities and feature views per read, that cost dominates.
118+
119+
Setting `client: glide` runs the same `HMGET` commands as one non-atomic [GLIDE](https://github.com/valkey-io/valkey-glide) batch. GLIDE has a Rust core, so the batch crosses the FFI boundary once and the fetch runs with the GIL released.
120+
121+
```yaml
122+
online_store:
123+
type: redis
124+
connection_string: "localhost:6379"
125+
client: glide
126+
```
127+
128+
Install the extra to use it:
129+
130+
```bash
131+
pip install 'feast[glide]'
132+
```
133+
134+
Notes:
135+
136+
* GLIDE reads the same keys and hash fields Feast already writes, so an existing store needs no migration and no rewrite of data.
137+
* The default stays `client: redis`. Writes and `get_online_features_async` keep using redis-py, so only the synchronous read path changes.
138+
* GLIDE supports `host:port`, `db`, `password`, `username`, `ssl`, `socket_timeout`, and `socket_connect_timeout` from `connection_string`. Other parameters are logged and ignored.
139+
* GLIDE has no Sentinel support, so `client: glide` cannot be combined with `redis_type: redis_sentinel`. Use `client: redis` for Sentinel deployments.
140+
141+
If `client: glide` is set without the extra installed, Feast raises an import error naming the extra to install.
142+
115143
### Write path: `skip_dedup` for bulk loads
116144

117145
By default, `online_write_batch()` checks existing timestamps before writing (to avoid overwriting newer data with older data). This requires two pipeline round trips per batch: one to read existing timestamps, one to write new values.
@@ -144,6 +172,7 @@ The Redis online store implements `online_write_batch_async()` using the async R
144172
| `key_ttl_seconds` | `null` | Redis `EXPIRE` TTL in seconds applied to the entity hash key after each write. Expires all feature views for that entity together. |
145173
| `full_scan_for_deletion` | `true` | When `true`, deleting or renaming a feature view scans Redis to remove its hash fields. Set `false` to skip deletion scans (faster `feast apply`, but leaves orphaned data). |
146174
| `skip_dedup` | `false` | When `true`, skips the existing-timestamp read before each write, halving write round trips. Suitable for initial bulk loads; may cause older values to overwrite newer ones under concurrent writers. |
175+
| `client` | `redis` | Read client: `redis` for redis-py, or `glide` to run the synchronous batched read as one non-atomic GLIDE batch. Requires `pip install 'feast[glide]'`. |
147176

148177
The full set of configuration options is available in [RedisOnlineStoreConfig](https://rtd.feast.dev/en/latest/#feast.infra.online_stores.redis.RedisOnlineStoreConfig).
149178

@@ -172,5 +201,6 @@ Below is a matrix indicating which functionality is supported by the Redis onlin
172201
| collocated by entity key | yes |
173202
| async batch writes | yes |
174203
| batched multi-feature-view reads (single pipeline) | yes |
204+
| optional GLIDE client for synchronous reads | yes |
175205

176206
To compare this set of functionality against other online stores, please see the full [functionality matrix](overview.md#functionality-matrix).

‎pyproject.toml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ gcp = [
8080
"fsspec>=2024.1.0",
8181
]
8282
ge = ["great_expectations>=0.15.41,<1"]
83+
glide = ["valkey-glide-sync>=2.5,<3"]
8384
go = ["cffi>=1.15.0"]
8485
grpcio = [
8586
"grpcio>=1.56.2",
@@ -166,7 +167,7 @@ test = [
166167
]
167168

168169
ci = [
169-
"feast[test, aws, azure, cassandra, clickhouse, couchbase, delta, docling, duckdb, elasticsearch, faiss, gcp, ge, go, grpcio, hazelcast, hbase, ibis, iceberg, image, k8s, mcp, milvus, mlflow, mongodb, mssql, mysql, openlineage, opentelemetry, oracle, scylladb, spark, trino, postgres, pytorch, qdrant, rag, ray, redis, singlestore, snowflake, sqlite_vec]",
170+
"feast[test, aws, azure, cassandra, clickhouse, couchbase, delta, docling, duckdb, elasticsearch, faiss, gcp, ge, glide, go, grpcio, hazelcast, hbase, ibis, iceberg, image, k8s, mcp, milvus, mlflow, mongodb, mssql, mysql, openlineage, opentelemetry, oracle, scylladb, spark, trino, postgres, pytorch, qdrant, rag, ray, redis, singlestore, snowflake, sqlite_vec]",
170171
"build",
171172
"virtualenv==20.23.0",
172173
"dbt-artifacts-parser",

‎sdk/python/feast/infra/online_stores/redis.py‎

Lines changed: 192 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
from pydantic import StrictStr
3333

3434
from feast import Entity, FeatureView, RepoConfig, utils
35+
from feast.errors import FeastExtrasDependencyImportError
3536
from feast.infra.key_encoding_utils import serialize_entity_key
3637
from feast.infra.online_stores.helpers import (
3738
_mmh3,
@@ -51,8 +52,6 @@
5152
from redis.cluster import ClusterNode, RedisCluster
5253
from redis.sentinel import Sentinel
5354
except ImportError as e:
54-
from feast.errors import FeastExtrasDependencyImportError
55-
5655
raise FeastExtrasDependencyImportError("redis", str(e))
5756

5857
logger = logging.getLogger(__name__)
@@ -71,6 +70,11 @@ class RedisType(str, Enum):
7170
redis_sentinel = "redis_sentinel"
7271

7372

73+
class RedisClient(str, Enum):
74+
redis = "redis"
75+
glide = "glide"
76+
77+
7478
class RedisOnlineStoreConfig(FeastConfigBaseModel):
7579
"""Online store config for Redis store"""
7680

@@ -80,6 +84,15 @@ class RedisOnlineStoreConfig(FeastConfigBaseModel):
8084
redis_type: RedisType = RedisType.redis
8185
"""Redis type: redis or redis_cluster"""
8286

87+
client: RedisClient = RedisClient.redis
88+
"""Redis client used for online reads: ``redis`` (redis-py, default) or ``glide``.
89+
90+
With ``glide``, the synchronous read paths (``online_read`` and the batched read
91+
behind ``get_online_features``) issue their HMGET commands as one non-atomic GLIDE
92+
batch, so the fetch runs off the GIL. Requires ``pip install 'feast[glide]'``.
93+
Writes and async reads always use redis-py. GLIDE has no Sentinel support, so
94+
``client: glide`` cannot be combined with ``redis_type: redis_sentinel``."""
95+
8396
sentinel_master: StrictStr = "mymaster"
8497
"""Sentinel's master name"""
8598

@@ -99,6 +112,115 @@ class RedisOnlineStoreConfig(FeastConfigBaseModel):
99112
This may cause older feature values to overwrite newer ones under concurrent writers."""
100113

101114

115+
def _seconds_to_millis(value: Any) -> Optional[int]:
116+
"""Convert a redis-py timeout in seconds to GLIDE's millisecond unit."""
117+
if value is None:
118+
return None
119+
return int(float(value) * 1000)
120+
121+
122+
def _load_glide_sync():
123+
"""Import ``valkey-glide-sync`` on demand.
124+
125+
The native extension is only loaded when ``client: glide`` is configured, so the
126+
redis extra keeps working for users who did not opt in.
127+
"""
128+
try:
129+
import glide_sync
130+
except ImportError as e:
131+
raise FeastExtrasDependencyImportError("glide", str(e)) from e
132+
return glide_sync
133+
134+
135+
_GLIDE_SUPPORTED_CONNECTION_PARAMS = frozenset(
136+
{"ssl", "db", "password", "username", "socket_timeout", "socket_connect_timeout"}
137+
)
138+
139+
140+
def _glide_client_config(glide_sync, online_store_config: RedisOnlineStoreConfig):
141+
"""Translate a Feast Redis connection string into a GLIDE client configuration."""
142+
if online_store_config.redis_type == RedisType.redis_sentinel:
143+
raise ValueError(
144+
"client: glide does not support redis_type: redis_sentinel. "
145+
"Use client: redis for Sentinel deployments."
146+
)
147+
148+
startup_nodes, params = RedisOnlineStore._parse_connection_string(
149+
online_store_config.connection_string
150+
)
151+
ignored = sorted(set(params) - _GLIDE_SUPPORTED_CONNECTION_PARAMS)
152+
if ignored:
153+
logger.warning(
154+
"client: glide ignores these Redis connection_string parameters: %s",
155+
", ".join(ignored),
156+
)
157+
158+
credentials = None
159+
password = params.get("password")
160+
username = params.get("username")
161+
if password is not None or username is not None:
162+
credentials = glide_sync.ServerCredentials(
163+
# _parse_connection_string json-decodes values, so a numeric password can
164+
# arrive as an int while GLIDE expects a str.
165+
password=None if password is None else str(password),
166+
username=None if username is None else str(username),
167+
)
168+
169+
connection_timeout = _seconds_to_millis(params.get("socket_connect_timeout"))
170+
kwargs: Dict[str, Any] = {
171+
"addresses": [
172+
glide_sync.NodeAddress(host=node["host"], port=int(node["port"]))
173+
for node in startup_nodes
174+
],
175+
"use_tls": bool(params.get("ssl", False)),
176+
"credentials": credentials,
177+
"request_timeout": _seconds_to_millis(params.get("socket_timeout")),
178+
"advanced_config": glide_sync.AdvancedGlideClientConfiguration(
179+
connection_timeout=connection_timeout
180+
)
181+
if connection_timeout is not None
182+
else None,
183+
}
184+
185+
if online_store_config.redis_type == RedisType.redis_cluster:
186+
# Cluster nodes do not support SELECT, so `db` is not passed along.
187+
return glide_sync.GlideClusterClientConfiguration(**kwargs)
188+
189+
if "db" in params:
190+
kwargs["database_id"] = int(params["db"])
191+
return glide_sync.GlideClientConfiguration(**kwargs)
192+
193+
194+
def _create_glide_client(glide_sync, online_store_config: RedisOnlineStoreConfig):
195+
"""Connect a GLIDE client built from the Redis online store config."""
196+
config = _glide_client_config(glide_sync, online_store_config)
197+
if online_store_config.redis_type == RedisType.redis_cluster:
198+
return glide_sync.GlideClusterClient.create(config)
199+
return glide_sync.GlideClient.create(config)
200+
201+
202+
def _glide_hmget_batch(
203+
glide_sync,
204+
client,
205+
commands: List[Tuple[bytes, List[Union[str, bytes]]]],
206+
) -> List[List[Optional[bytes]]]:
207+
"""Run one HMGET per ``(key, fields)`` pair in a single non-atomic GLIDE batch.
208+
209+
Replies keep the order of ``commands``, so callers can slice them exactly like a
210+
redis-py pipeline result.
211+
"""
212+
batch = (
213+
glide_sync.ClusterBatch(is_atomic=False)
214+
if isinstance(client, glide_sync.GlideClusterClient)
215+
else glide_sync.Batch(is_atomic=False)
216+
)
217+
for key, fields in commands:
218+
batch.hmget(key, fields)
219+
220+
result = client.exec(batch, raise_on_error=True)
221+
return list(result) if result else []
222+
223+
102224
class RedisOnlineStore(OnlineStore):
103225
"""
104226
Redis implementation of the online store interface.
@@ -114,6 +236,10 @@ class RedisOnlineStore(OnlineStore):
114236
_client_async: Optional[Union[redis_asyncio.Redis, redis_asyncio.RedisCluster]] = (
115237
None
116238
)
239+
# Only set when `client: glide` is configured, so the native extension stays
240+
# untouched for users who never opt in.
241+
_glide_client: Optional[Any] = None
242+
_glide_sync: Optional[Any] = None
117243

118244
@property
119245
def async_supported(self) -> SupportedAsyncMethods:
@@ -225,6 +351,12 @@ def teardown(
225351
for join_keys in join_keys_to_delete:
226352
self.delete_entity_values(config, list(join_keys))
227353

354+
# Release the native GLIDE client, if one was ever opened, so teardown does
355+
# not leave its Rust-side resources and file descriptors behind.
356+
if self._glide_client:
357+
self._glide_client.close()
358+
self._glide_client = None
359+
228360
@staticmethod
229361
def _parse_connection_string(connection_string: str):
230362
"""
@@ -305,6 +437,20 @@ async def _get_client_async(self, online_store_config: RedisOnlineStoreConfig):
305437
self._client_async = redis_asyncio.Redis(**kwargs)
306438
return self._client_async
307439

440+
def _get_glide_sync(self):
441+
"""Returns the cached ``glide_sync`` module, importing it on first use."""
442+
if self._glide_sync is None:
443+
self._glide_sync = _load_glide_sync()
444+
return self._glide_sync
445+
446+
def _get_glide_client(self, online_store_config: RedisOnlineStoreConfig):
447+
"""Creates and caches a GLIDE client built from the online store config."""
448+
if not self._glide_client:
449+
self._glide_client = _create_glide_client(
450+
self._get_glide_sync(), online_store_config
451+
)
452+
return self._glide_client
453+
308454
def online_write_batch(
309455
self,
310456
config: RepoConfig,
@@ -538,12 +684,14 @@ def _generate_hset_keys_for_features(
538684
feature_view: FeatureView,
539685
requested_features: Optional[List[str]] = None,
540686
fv_name_override: Optional[str] = None,
541-
) -> Tuple[List[str], List[str]]:
687+
) -> Tuple[List[str], List[Union[str, bytes]]]:
542688
if not requested_features:
543689
requested_features = [f.name for f in feature_view.features]
544690

545691
fv_name = fv_name_override or feature_view.name
546-
hset_keys = [_mmh3(f"{fv_name}:{k}") for k in requested_features]
692+
hset_keys: List[Union[str, bytes]] = [
693+
_mmh3(f"{fv_name}:{k}") for k in requested_features
694+
]
547695

548696
ts_key = f"_ts:{fv_name}"
549697
hset_keys.append(ts_key)
@@ -553,7 +701,7 @@ def _generate_hset_keys_for_features(
553701

554702
def _convert_redis_values_to_protobuf(
555703
self,
556-
redis_values: List[List[ByteString]],
704+
redis_values: Sequence[Sequence[Optional[ByteString]]],
557705
feature_view: str,
558706
requested_features: List[str],
559707
) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
@@ -562,17 +710,40 @@ def _convert_redis_values_to_protobuf(
562710
for values in redis_values
563711
]
564712

713+
def _read_hash_fields(
714+
self,
715+
config: RepoConfig,
716+
commands: List[Tuple[bytes, List[Union[str, bytes]]]],
717+
) -> List[List[Optional[ByteString]]]:
718+
"""Fetch one HMGET reply per ``(key, fields)`` pair, in the given order.
719+
720+
Defaults to a single redis-py pipeline. When ``client: glide`` is configured
721+
the same commands run as one non-atomic GLIDE batch, so the fetch runs off the
722+
GIL instead of in Python per command.
723+
"""
724+
online_store_config = config.online_store
725+
assert isinstance(online_store_config, RedisOnlineStoreConfig)
726+
727+
if online_store_config.client == RedisClient.glide:
728+
return _glide_hmget_batch(
729+
self._get_glide_sync(),
730+
self._get_glide_client(online_store_config),
731+
commands,
732+
)
733+
734+
client = self._get_client(online_store_config)
735+
with client.pipeline(transaction=False) as pipe:
736+
for redis_key, fields in commands:
737+
pipe.hmget(redis_key, fields)
738+
return pipe.execute()
739+
565740
def online_read(
566741
self,
567742
config: RepoConfig,
568743
table: FeatureView,
569744
entity_keys: List[EntityKeyProto],
570745
requested_features: Optional[List[str]] = None,
571746
) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
572-
online_store_config = config.online_store
573-
assert isinstance(online_store_config, RedisOnlineStoreConfig)
574-
575-
client = self._get_client(online_store_config)
576747
feature_view = table
577748
fv_name = _versioned_fv_name(table, config)
578749

@@ -581,11 +752,9 @@ def online_read(
581752
)
582753
keys = self._generate_redis_keys_for_entities(config, entity_keys)
583754

584-
with client.pipeline(transaction=False) as pipe:
585-
for redis_key_bin in keys:
586-
pipe.hmget(redis_key_bin, hset_keys)
587-
588-
redis_values = pipe.execute()
755+
redis_values = self._read_hash_fields(
756+
config, [(redis_key, hset_keys) for redis_key in keys]
757+
)
589758

590759
return self._convert_redis_values_to_protobuf(
591760
redis_values, fv_name, requested_features
@@ -648,12 +817,14 @@ def _read_features_per_fv(
648817
)
649818

650819
if work_items:
651-
client = self._get_client(config.online_store)
652-
with client.pipeline(transaction=False) as pipe:
653-
for _, _, _, hset_keys, redis_keys, _, _ in work_items:
654-
for redis_key in redis_keys:
655-
pipe.hmget(redis_key, hset_keys)
656-
all_results = pipe.execute()
820+
all_results = self._read_hash_fields(
821+
config,
822+
[
823+
(redis_key, hset_keys)
824+
for _, _, _, hset_keys, redis_keys, _, _ in work_items
825+
for redis_key in redis_keys
826+
],
827+
)
657828

658829
offset = 0
659830
for (
@@ -751,7 +922,7 @@ async def _read_features_per_fv_async(
751922

752923
def _get_features_for_entity(
753924
self,
754-
values: List[ByteString],
925+
values: Sequence[Optional[ByteString]],
755926
feature_view: str,
756927
requested_features: List[str],
757928
) -> Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]:

0 commit comments

Comments
 (0)