fix: Preserve event-time ordering within Redis online_write_batch - #6656
Conversation
franciscojavierarceo
left a comment
There was a problem hiding this comment.
The per-key running maximum is applied consistently in both sync and async write paths, and the added cases cover descending, ascending, unordered, and pre-existing stored timestamps. I found no blocking issue in the remote diff.
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #6656 +/- ##
==========================================
+ Coverage 45.98% 46.06% +0.07%
==========================================
Files 414 414
Lines 50037 50045 +8
Branches 7147 7147
==========================================
+ Hits 23012 23051 +39
+ Misses 25413 25380 -33
- Partials 1612 1614 +2
... and 1 file with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
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 feast-dev#5163 Signed-off-by: adarshsm <24850536+adarshsm@users.noreply.github.com>
a1ebc75 to
edc9781
Compare
What this PR does / why we need it:
online_write_batchperforms its staleness check against a snapshot that is readbefore any write is queued:
hmgetthe stored event timestamp for every row's entity key, thenpipe.execute()once.<=the stored one,otherwise queue an
hset.Because the read phase completes before the first write is queued, rows in the same batch
that share an entity key all compare against the same pre-batch value. None of them can
see the others. When nothing is stored yet,
prev_event_timeisNonefor all of them,the guard is skipped entirely, and every row gets an
hsetqueued. Redis appliespipelined commands in order, so the last row in list order wins — regardless of its
event time.
Pushing one entity key with timestamps in descending order therefore leaves the oldest
value in the store:
No error and no warning — just a silently incorrect feature value.
The fix
Track the latest event timestamp queued so far per entity key within the batch, and
compare each row against whichever is newer: the stored timestamp, or an earlier row of
the same batch.
redis_key_binis already computed in the read phase and uniquely identifies(project, entity_key), so it was free to reuse as the dedup key. The guard had to moveout of the
if prev_event_time:block, withprev_total_nanosinitialised to0so themax()works when nothing is stored yet — that's the structural part of the diff.online_write_batch_asynchad the identical defect and gets the same fix; fixing only thesync path would have left half the bug in place.
The
skip_dedupfast path is deliberately untouched. Its own comment says it is "suitablefor initial loads or append-only pipelines where out-of-order writes are not a concern," so
losing ordering there is the documented tradeoff of enabling it.
One design note worth your input
The issue suggests sorting the batch by timestamp, or reducing to one row per key. Either
would work, but both change behaviour for batches that are already correct — they alter
how many writes get queued in the ascending case and shift the
progress()accounting.I went with the running-max approach because it repairs only the broken case and leaves
every already-working case byte-identical. Happy to switch to sort-or-reduce if you'd
rather have the write-amplification reduction too; it's a small change from here.
Behaviour that is intentionally preserved:
<="older or sameinstant" semantics.
truthy
prev_total_nanos, andmax(0, ...)preserves that.Note on the labels
The
wontfixlabel on #5163 was applied bystale[bot]as its configuredstaleLabel,not by a maintainer declining the report — the timeline shows the bot's comment as the only
activity on the issue. Flagging it so it doesn't read as a prior decision.
kind/bugandpriority/p2are the defaults frombug_report.md. If this is out of scope for otherreasons, I'm glad to hear it.
Which issue(s) this PR fixes:
Fixes #5163
Checks
git commit -s)Testing Strategy
Misc
Five tests added to
sdk/python/tests/unit/infra/online_store/test_redis.py, reusing theexisting
MagicMockpipeline idiom, so no Redis server or Docker is needed:keeps_latest_event_time_within_batch[descending]keeps_latest_event_time_within_batch[unordered]keeps_latest_event_time_within_batch[ascending]async_keeps_latest_event_time_within_batchstill_skips_rows_older_than_stored_valueThe two that pass before the fix are the load-bearing ones — they're there to show the
change repairs the broken orderings without disturbing the ascending case or the existing
staleness guard.
The pre-fix failure is
assert b'\x18\n' == b'\x18\x1e', i.e.int32_val=10where 30 wasexpected — the reporter's "returns 10 instead of 30", reproduced deterministically.
test_redis.pygoes 18 → 23 passing.ruff check,ruff format --checkandmypy feast/infra/online_stores/redis.pyare all clean.I ran unit tests only, not the integration suite — it needs cloud credentials I don't have.
Happy to add an integration test if you'd like one for this path.
One small correction to the issue for the record: it describes the code as skipping
"records with timestamps older than what's already been processed in the current batch."
Nothing intra-batch is tracked today, which is exactly why the bug exists — but the symptom
as reported is accurate.