Visitar URL original
fix: Preserve event-time ordering within Redis online_write_batch · feast-dev/feast@40fb788 · GitHub
Skip to content

Commit 40fb788

Browse files
adarshsmntkathole
authored andcommitted
fix: Preserve event-time ordering within Redis online_write_batch
online_write_batch reads every existing event timestamp in a single pipeline before it queues any write, so rows in the same batch that share an entity key all compared against the same pre-batch snapshot. None of them could see the others, so every row passed the staleness check and Redis applied the queued writes in list order — the last row won regardless of its event time. A batch of [t3, t2, t1] for one entity key left the value belonging to t1 in the store. Track the latest event timestamp queued so far for each entity key and compare each row against whichever is newer: the stored timestamp, or an earlier row of the same batch. Batches that already arrived in ascending order keep their existing behaviour, and rows older than the value already stored are still skipped. online_write_batch_async had the identical defect and gets the same fix. The skip_dedup fast path is deliberately left alone, since it documents that out-of-order writes are not a concern when it is enabled. Fixes #5163 Signed-off-by: adarshsm <24850536+adarshsm@users.noreply.github.com>
1 parent adb8c1c commit 40fb788

2 files changed

Lines changed: 172 additions & 10 deletions

File tree

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

Lines changed: 31 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -370,6 +370,13 @@ def online_write_batch(
370370
# flattening the list of lists. `hmget` does the lookup assuming a list of keys in the key bin
371371
prev_event_timestamps = [i[0] for i in prev_event_timestamps]
372372

373+
# Latest event timestamp queued so far in this batch, per entity key. The
374+
# reads above all happened before any write, so without this rows sharing
375+
# an entity key would each compare against the same pre-batch value, all
376+
# pass the staleness check, and the last one queued would win regardless
377+
# of its event time.
378+
batch_latest_nanos: Dict[bytes, int] = {}
379+
373380
for redis_key_bin, prev_event_time, (_, values, timestamp, _) in zip(
374381
keys, prev_event_timestamps, data
375382
):
@@ -381,15 +388,21 @@ def online_write_batch(
381388
# New timestamp in nanoseconds
382389
new_total_nanos = ts.seconds * 1_000_000_000 + ts.nanos
383390
# Compare against existing timestamp (nanosecond precision)
391+
prev_total_nanos = 0
384392
if prev_event_time:
385393
prev_ts = Timestamp()
386394
prev_ts.ParseFromString(prev_event_time)
387395
prev_total_nanos = prev_ts.seconds * 1_000_000_000 + prev_ts.nanos
388-
# Skip only if older OR exact same instant
389-
if prev_total_nanos and new_total_nanos <= prev_total_nanos:
390-
if progress:
391-
progress(1)
392-
continue
396+
# Whichever is newer: the stored value or an earlier row of this batch
397+
latest_seen_nanos = max(
398+
prev_total_nanos, batch_latest_nanos.get(redis_key_bin, 0)
399+
)
400+
# Skip only if older OR exact same instant
401+
if latest_seen_nanos and new_total_nanos <= latest_seen_nanos:
402+
if progress:
403+
progress(1)
404+
continue
405+
batch_latest_nanos[redis_key_bin] = new_total_nanos
393406
# Store full timestamp (seconds + nanos)
394407
entity_hset = {ts_key: ts.SerializeToString()}
395408

@@ -466,6 +479,10 @@ async def online_write_batch_async(
466479

467480
prev_event_timestamps = [i[0] for i in prev_event_timestamps]
468481

482+
# See online_write_batch: guards against rows in this batch that share an
483+
# entity key overwriting each other out of event-time order.
484+
batch_latest_nanos: Dict[bytes, int] = {}
485+
469486
async with client.pipeline(transaction=False) as pipe:
470487
for redis_key_bin, prev_event_time, (_, values, timestamp, _) in zip(
471488
keys, prev_event_timestamps, data
@@ -475,14 +492,19 @@ async def online_write_batch_async(
475492
ts.FromDatetime(aware_ts)
476493
new_total_nanos = ts.seconds * 1_000_000_000 + ts.nanos
477494

495+
prev_total_nanos = 0
478496
if prev_event_time:
479497
prev_ts = Timestamp()
480498
prev_ts.ParseFromString(prev_event_time)
481499
prev_total_nanos = prev_ts.seconds * 1_000_000_000 + prev_ts.nanos
482-
if prev_total_nanos and new_total_nanos <= prev_total_nanos:
483-
if progress:
484-
progress(1)
485-
continue
500+
latest_seen_nanos = max(
501+
prev_total_nanos, batch_latest_nanos.get(redis_key_bin, 0)
502+
)
503+
if latest_seen_nanos and new_total_nanos <= latest_seen_nanos:
504+
if progress:
505+
progress(1)
506+
continue
507+
batch_latest_nanos[redis_key_bin] = new_total_nanos
486508

487509
entity_hset = {ts_key: ts.SerializeToString()}
488510
for feature_name, val in values.items():

‎sdk/python/tests/unit/infra/online_store/test_redis.py‎

Lines changed: 141 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
import asyncio
2-
from datetime import datetime, timezone
2+
from datetime import datetime, timedelta, timezone
33
from unittest.mock import AsyncMock, MagicMock, patch
44

55
import pytest
66
from google.protobuf.timestamp_pb2 import Timestamp
77

88
from feast import Entity, FeatureView, Field, FileSource, RepoConfig
9+
from feast.infra.online_stores.helpers import _mmh3
910
from feast.infra.online_stores.redis import RedisOnlineStore, RedisOnlineStoreConfig
1011
from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
1112
from feast.protos.feast.types.Value_pb2 import Value as ValueProto
@@ -426,6 +427,145 @@ def test_online_write_batch_with_dedup_uses_two_pipelines(
426427
pipe.hmget.assert_called_once()
427428

428429

430+
def _dedup_config():
431+
return RepoConfig(
432+
provider="local",
433+
project="test",
434+
entity_key_serialization_version=3,
435+
registry="dummy_registry.db",
436+
online_store=RedisOnlineStoreConfig(), # default: skip_dedup=False
437+
)
438+
439+
440+
def _single_entity_batch(timestamps_and_values):
441+
"""Build a write batch for one entity key, one row per (timestamp, value) pair."""
442+
entity_key = EntityKeyProto(
443+
join_keys=["entity"], entity_values=[ValueProto(int32_val=1)]
444+
)
445+
return [
446+
(entity_key, {"feature_10": ValueProto(int32_val=value)}, timestamp, None)
447+
for timestamp, value in timestamps_and_values
448+
]
449+
450+
451+
def _written_feature_values(hset_calls, feature_view_name, feature_name):
452+
"""Extract the serialized feature value from each queued hset call."""
453+
f_key = _mmh3(f"{feature_view_name}:{feature_name}")
454+
return [call.kwargs["mapping"][f_key] for call in hset_calls]
455+
456+
457+
@pytest.mark.parametrize(
458+
"order",
459+
["descending", "ascending", "unordered"],
460+
ids=["descending", "ascending", "unordered"],
461+
)
462+
def test_online_write_batch_keeps_latest_event_time_within_batch(
463+
redis_online_store: RedisOnlineStore, feature_view, order
464+
):
465+
"""A batch containing several rows for one entity key must leave the value
466+
belonging to the latest event timestamp in the store, whatever order the rows
467+
arrive in.
468+
469+
Regression test for #5163: previous timestamps were read once up front, so rows
470+
sharing an entity key all compared against the same pre-batch snapshot and could
471+
not see each other. Every row passed the staleness guard and the last row queued
472+
won, so a reverse-chronological batch stored the *oldest* value.
473+
"""
474+
t1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
475+
t2 = t1 + timedelta(seconds=5)
476+
t3 = t2 + timedelta(seconds=5)
477+
478+
rows = {
479+
"descending": [(t3, 30), (t2, 20), (t1, 10)],
480+
"ascending": [(t1, 10), (t2, 20), (t3, 30)],
481+
"unordered": [(t2, 20), (t3, 30), (t1, 10)],
482+
}[order]
483+
data = _single_entity_batch(rows)
484+
485+
mock_client = MagicMock()
486+
pipe = MagicMock()
487+
pipe.__enter__ = MagicMock(return_value=pipe)
488+
pipe.__exit__ = MagicMock(return_value=False)
489+
# No pre-existing value: one hmget result per row, each holding None.
490+
pipe.execute.side_effect = [[[None]] * len(data), []]
491+
mock_client.pipeline.return_value = pipe
492+
493+
with patch.object(redis_online_store, "_get_client", return_value=mock_client):
494+
redis_online_store.online_write_batch(
495+
_dedup_config(), feature_view, data, progress=None
496+
)
497+
498+
written = _written_feature_values(
499+
pipe.hset.call_args_list, feature_view.name, "feature_10"
500+
)
501+
assert written, "no write was queued for the batch"
502+
# Redis applies queued commands in order, so the surviving value is the last one.
503+
assert written[-1] == ValueProto(int32_val=30).SerializeToString()
504+
505+
506+
def test_online_write_batch_async_keeps_latest_event_time_within_batch(
507+
redis_online_store: RedisOnlineStore, feature_view
508+
):
509+
"""online_write_batch_async must honour the same intra-batch ordering guarantee
510+
as the sync path (#5163)."""
511+
t1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
512+
t2 = t1 + timedelta(seconds=5)
513+
t3 = t2 + timedelta(seconds=5)
514+
data = _single_entity_batch([(t3, 30), (t2, 20), (t1, 10)])
515+
516+
async_pipe = AsyncMock()
517+
async_pipe.__aenter__ = AsyncMock(return_value=async_pipe)
518+
async_pipe.__aexit__ = AsyncMock(return_value=False)
519+
async_pipe.execute = AsyncMock(side_effect=[[[None]] * len(data), []])
520+
async_pipe.hset = MagicMock()
521+
522+
mock_async_client = AsyncMock()
523+
mock_async_client.pipeline = MagicMock(return_value=async_pipe)
524+
525+
async def _run():
526+
with patch.object(
527+
redis_online_store,
528+
"_get_client_async",
529+
AsyncMock(return_value=mock_async_client),
530+
):
531+
await redis_online_store.online_write_batch_async(
532+
_dedup_config(), feature_view, data, progress=None
533+
)
534+
535+
asyncio.run(_run())
536+
537+
written = _written_feature_values(
538+
async_pipe.hset.call_args_list, feature_view.name, "feature_10"
539+
)
540+
assert written, "no write was queued for the batch"
541+
assert written[-1] == ValueProto(int32_val=30).SerializeToString()
542+
543+
544+
def test_online_write_batch_still_skips_rows_older_than_stored_value(
545+
redis_online_store: RedisOnlineStore, feature_view
546+
):
547+
"""The pre-existing staleness guard must keep working: a row older than the value
548+
already in Redis is dropped rather than written."""
549+
stored_ts = Timestamp()
550+
stored_ts.FromDatetime(datetime(2024, 1, 1, 12, 0, 10, tzinfo=timezone.utc))
551+
older = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
552+
data = _single_entity_batch([(older, 10)])
553+
554+
mock_client = MagicMock()
555+
pipe = MagicMock()
556+
pipe.__enter__ = MagicMock(return_value=pipe)
557+
pipe.__exit__ = MagicMock(return_value=False)
558+
pipe.execute.side_effect = [[[stored_ts.SerializeToString()]], []]
559+
mock_client.pipeline.return_value = pipe
560+
561+
with patch.object(redis_online_store, "_get_client", return_value=mock_client):
562+
redis_online_store.online_write_batch(
563+
_dedup_config(), feature_view, data, progress=None
564+
)
565+
566+
pipe.hset.assert_not_called()
567+
568+
429569
def test_online_write_batch_async_skip_dedup_single_pipeline(
430570
redis_online_store: RedisOnlineStore, feature_view
431571
):

0 commit comments

Comments
 (0)