fix(cache): stop evicted entries outliving the budget - #45
Conversation
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // there: demote it by flipping the index entry rather than | ||
| // re-serialising it into a hybrid whose backing would have to be | ||
| // written all over again. | ||
| let outcome = match (to_squeeze_batch.as_ref(), self.disk_copy(&to_squeeze)) { |
There was a problem hiding this comment.
Nothing invalidates disk_copies when a caller overwrites an entry. This arm then flips the index to a disk stub that points at the previous value.
Failure scenario:
cache.insert(id, v1)storesMemoryArrow(v1).- Memory pressure demotes
idtoDiskLiquidand recordsdisk_copies[id] = Liquid/N. cache.insert(id, v2)replaces the index entry withMemoryArrow(v2). No write reaches the store, so the store still holds v1.- Memory pressure transcodes v2 to
MemoryLiquid. - This arm matches and inserts
disk_liquid(dt, N)withbytes_to_write: None. cache.get(&id)reads the store and returns v1.
The shortcuts at line 287 and line 365 have the same defect.
Overwrite is a supported operation. try_insert at line 498 handles the existing-entry case explicitly, and test_basic_cache_operations at line 1166 replaces entry 1 with an array of a different length. LiquidCache::insert is also public API.
Fix: drop the disk_copies record for caller-supplied inserts. Insert::run in src/core/src/cache/builders.rs:206 is the boundary, because insert_inner is shared with maybe_hydrate and hydration must keep the record.
There was a problem hiding this comment.
Fixed in 3ae1a4e. Insert::run now calls supersede_disk_copy before insert_inner: if the index still holds the stub, the whole entry goes through remove_disk_entry; if it was hydrated, the record is dropped, the store object removed and its reservation released. Hydration is unaffected since it enters through insert_inner. Two tests cover the stub and the hydrated case, overwrite_of_disk_stub_invalidates_disk_copy and overwrite_of_hydrated_entry_invalidates_disk_copy (insert, demote, overwrite with a different-length array, demote, read back the new value, and check the old reservation is gone). Both fail on the previous head.
While writing them I found a second reachable defect in the same code: flush_all_to_disk writes an arrow entry as Arrow IPC but write_batch_to_disk recorded the copy as Liquid, so a flushed, hydrated, then transcoded entry was flipped to a liquid stub over arrow bytes and the read panicked in the IPC reader. Fixed in the same commit, covered by flushed_arrow_copy_is_not_reused_as_liquid.
| bytes, | ||
| }), | ||
| ) => SqueezeOutcome::Replace { | ||
| entry: CacheEntry::disk_liquid(liquid.original_arrow_data_type(), bytes), |
There was a problem hiding this comment.
nit: this arm bypasses the configured SqueezePolicy for every rehydrated liquid entry (not blocking).
The default policy is TranscodeSqueezeEvict. That policy turns MemoryLiquid into MemorySqueezedLiquid (src/core/src/cache/policies/squeeze.rs:131), which keeps a compact in-memory form for date32 and variant pushdown. Once an entry has been demoted to DiskLiquid and rehydrated, disk_copy returns Some, so the squeezed tier is never reached again for that entry.
Consider extending reuse_disk_copy to also drop the write for a MemorySqueezedLiquid outcome whose disk_backing() size and kind match the recorded copy. That avoids the rewrite without changing the entry shape the policy chose.
There was a problem hiding this comment.
Done in 3ae1a4e. The shortcut is now gated on squeeze_hint.is_none(): no LiquidArray::squeeze implementation produces a squeezed form without a hint, so unhinted the shortcut arrives at the same stub every policy would, minus the re-serialisation. With a hint the policy runs, and reuse_disk_copy now also handles a MemorySqueezedLiquid outcome, dropping the write when disk_backing() matches the recorded copy in kind and length. Covered by rehydrated_hinted_entry_returns_to_squeezed_tier_without_rewrite, which asserts the squeezed tier is reached with zero writes and the disk budget unchanged.
| }, | ||
| _ => DiskKind::Liquid, | ||
| }; | ||
| self.disk_copies |
There was a problem hiding this comment.
nit: release the previous copy's bytes when this call overwrites an existing disk object (not blocking).
try_reserve_disk(len) at line 910 reserves on every call. Overwriting a key that already has a DiskCopy holds the old reservation forever, so the disk tally exceeds the indexed on-disk bytes and the tier evicts real entries early.
A kind change reaches this path. An entry with an Arrow copy that later writes liquid bytes fails the copy.kind != stub.kind check in reuse_disk_copy, so the write proceeds. The new budget_disk_bytes == index_disk_bytes assertion in memory_footprint.rs passes only because that test never changes kind.
The map now carries the old size, so the fix is local: call self.budget.release_disk(old.bytes) when disk_copies already holds a record for entry_id.
There was a problem hiding this comment.
Done in 3ae1a4e: write_batch_to_disk releases the previous record's bytes when the put replaces the object under the key. The kind-change path you describe is exactly what flushed_arrow_copy_is_not_reused_as_liquid exercises now, and it asserts the disk budget equals the stub's bytes afterwards.
| Some(batch) | ||
| // A slot emptied by a concurrent remove reads as a miss, exactly as if | ||
| // the remove had won the race outright. | ||
| self.art.get(*entry_id, &guard)?.load() |
There was a problem hiding this comment.
nit: the comment covers the remove race but not the replace race (not blocking).
ArtIndex::insert empties the old slot at line 76. A reader that already loaded the old slot pointer gets None, although the key is present with a new value. That is not equivalent to a remove winning, because the entry is still cached.
Two consequences follow. is_cached reports false for a cached entry. In try_insert (src/core/src/cache/core.rs:498) the None branch calls try_reserve_memory instead of try_update_memory_usage, so the replaced entry's bytes are never released from the memory tally.
Consider retrying the ART lookup once when load() returns None, and updating the comment to name the replace case.
There was a problem hiding this comment.
Done in 7cb9f7c: ArtIndex::get re-runs the ART lookup once when the loaded slot is empty, and the comment now names the replace case.
There was a problem hiding this comment.
Review
Blocking Issues
src/core/src/cache/core.rs:624— thedisk_copiesrecord is never invalidated when a caller overwrites an entry. A later demotion flips the index to a disk stub that points at the previous value, sogetreturns stale data. The shortcuts at line 287 and line 365 share the defect. Details are in the inline comment.
Action Required
Invalidate the disk_copies record for caller-supplied inserts. Insert::run in src/core/src/cache/builders.rs:206 is the right boundary, because insert_inner is shared with maybe_hydrate and hydration must keep the record. Add a test that inserts an entry, demotes it to disk, overwrites it with different data, demotes it again, and reads the new value back.
Notes
Three non-blocking comments are inline: a SqueezePolicy bypass for rehydrated entries, a disk reservation that is never released on overwrite, and a replace race in ArtIndex::get.
CI had not reported any result when this review ran, so no check outcome is claimed here.
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
📊 Benchmark ComparisonCurrent:
Compared Liquid vs DataFusionDefault on the same runner |
`ArtIndex::insert` empties the replaced slot, so a reader that had already loaded the old slot saw `None` for a key that is still present. `is_cached` then reported false and `try_insert` reserved fresh memory for the new value instead of releasing the replaced one. Look the key up once more when the slot is empty.
A caller overwriting an entry left the disk-copy record pointing at the previous value, so a later demotion flipped the index to a stub over stale bytes and `get` returned them. `Insert` now removes the object and its reservation before inserting; hydration, which shares `insert_inner`, keeps the record because the bytes are still current. Also in `write_batch_to_disk`: a flushed arrow entry was recorded as a liquid copy, so a hydrated-then-transcoded entry could be demoted to a liquid stub over Arrow IPC bytes; and a put that replaces the object under a key now releases the previous copy's reservation. The hydrated-liquid shortcut in `squeeze_victim_inner` now applies only without a squeeze hint (no `LiquidArray::squeeze` produces a squeezed form unhinted); with one the policy runs, so the entry can return to the squeezed tier, and `reuse_disk_copy` drops the write when the squeezed form's backing is the copy already on disk.
| let Some(copy) = self.disk_copies.lock().unwrap().remove(&entry_id) else { | ||
| return; | ||
| }; | ||
| self.store |
There was a problem hiding this comment.
This branch deletes the store object of a MemorySqueezedLiquid entry and leaves that entry in the index. A squeezed entry reads its full value back through that object, so the entry becomes unreadable.
Failure scenario:
- Entry E sits in the squeezed tier.
disk_copies[E]records the backing and the store holds it. cache.insert(E, v2)reaches this branch. The object is removed and the reservation is released.insert_innercannot fit v2 in memory, andfind_memory_victimreturns nothing.write_in_memory_batch_to_diskfinds a full disk tier with no disk victim and returnsCacheFull.insertreturnsErr(CacheFull), so the index still holds the squeezed entry for E.cache.get(&E)reads the backing,store.getfails, andsrc/core/src/liquid_array/squeezed_date32_array.rs:241panics on.expect("read squeezed backing").
A concurrent reader of E hits the same panic during the window between step 2 and the new value reaching the index.
Fix: remove the index entry for a MemorySqueezedLiquid entry too, as the stub branch above already does. drop_memory_entry releases the memory tally for that entry.
There was a problem hiding this comment.
Fixed in ad483ad. supersede_disk_copy now sends a MemorySqueezedLiquid entry through drop_memory_entry (index entry out, memory tally released) before the object goes, the same way a stub goes through remove_disk_entry. Covered by overwrite_of_squeezed_entry_that_fails_to_insert_leaves_no_entry: a squeezed entry with the disk tier sized to exactly its backing, a NoVictims policy, and an oversized overwrite that returns CacheFull; afterwards the entry is absent, get is None, and both tallies are zero. Fails on the previous head.
| kind: DiskKind::Liquid, | ||
| bytes, | ||
| }), | ||
| None, |
There was a problem hiding this comment.
nit: LiquidFloatArray::squeeze ignores the hint, so this arm skips the squeezed tier for float entries (not blocking).
src/core/src/liquid_array/float_array.rs:341 names the parameter _expression_hint and returns Some for any bit width of 8 or more. An unhinted float entry therefore does have a squeezed form. This arm flips such an entry straight to a disk stub, so TranscodeSqueezeEvict never halves its bit width again after the first disk round trip. Float columns are common in the TPC-H tables this repo benchmarks.
Fix the comment above, or add a LiquidArray method that reports whether an unhinted squeeze exists and gate this arm on it.
There was a problem hiding this comment.
Good catch, the claim was wrong for floats. Rather than gate the arm on a new trait method I removed the shortcut in ad483ad: the policy always runs and reuse_disk_copy drops the rewrite when the chosen form is backed by the copy already on disk, so a rehydrated float entry halves its bit width again and an unhinted string entry still demotes without a write. The only cost is the serialisation the policy produces for the outcome, which is then discarded.
| .remove(&entry_id_to_key(&entry_id)) | ||
| .await | ||
| .expect("disk remove failed"); | ||
| self.disk_copies.lock().unwrap().remove(&entry_id); |
There was a problem hiding this comment.
nit: drop_memory_entry drops no disk-copy record, unlike this line (not blocking).
flush_all_to_disk calls drop_memory_entry when write_batch_to_disk returns CacheFull (line 294 and line 324). For an entry that holds a disk copy, the index entry then disappears while disk_copies keeps the record, the store keeps the object, and the budget keeps the reservation. No later path removes that object, so the disk tier stays smaller than configured, and the new budget_disk_bytes == index_disk_bytes assertion in src/core/tests/memory_footprint.rs no longer holds.
Remove the record in drop_memory_entry, delete the store object, and release the recorded bytes.
There was a problem hiding this comment.
Done in ad483ad: drop_memory_entry now calls the same discard_disk_copy that overwrite uses (record out, store object removed, reservation released). Covered by flush_dropping_hydrated_entry_releases_its_disk_copy: a flushed then hydrated arrow entry, a disk tier holding exactly its copy, and a second flush that cannot rewrite it; the entry is dropped and the disk tally returns to zero. Fails on the previous head.
There was a problem hiding this comment.
Review
The prior blocking finding is fixed for a disk stub and for a hydrated liquid entry. supersede_disk_copy opens one new hole for a squeezed entry.
Blocking Issues
src/core/src/cache/core.rs:503—supersede_disk_copydeletes the store object of aMemorySqueezedLiquidentry and keeps that entry in the index. A squeezed entry reads its value back through that object. Wheninsert_innerthen returnsCacheFull, the entry stays in the index over a deleted object, and every later read panics atsrc/core/src/liquid_array/squeezed_date32_array.rs:241.
Action Required
- In
supersede_disk_copy, remove the index entry for aMemorySqueezedLiquidentry as well, the way the stub branch does. Release its memory tally throughdrop_memory_entry. - Add a test that overwrites an entry which sits in the squeezed tier, then reads that entry.
Two non-blocking comments sit inline on the float squeeze arm and on drop_memory_entry.
CI checks were queued or in progress when this review started, so no check result is reported here.
There was a problem hiding this comment.
Review
Blocking Issues
src/core/src/cache/core.rs:500-507 — supersede_disk_copy deletes the store object of a MemorySqueezedLiquid entry and leaves that entry in the index. The finding is unchanged from thread 3926032595. The diff since the last review touches only snapshot files.
A squeezed entry reads its full value back through that store object, so the entry becomes unreadable.
Failure scenario:
- Entry E sits in the squeezed tier.
disk_copies[E]records the backing and the store holds it. cache.insert(E, v2)reaches line 500. The object is removed and the reservation is released.insert_inner(line 406) cannot fit v2 in memory andfind_memory_victimreturns nothing.write_in_memory_batch_to_diskfinds a full disk tier with no disk victim and returnsCacheFull.insertreturnsErr(CacheFull), so the index still holds the squeezed entry for E.cache.get(&E)reads the backing,store.getfails, andsrc/core/src/liquid_array/squeezed_date32_array.rs:241panics on.expect("read squeezed backing").
A second path reaches an unreadable entry without any error return. TranscodeSqueezeEvict::squeeze at src/core/src/cache/policies/squeeze.rs:151 turns a squeezed entry into a disk stub over the same object, with bytes_to_write: None. If E is picked as a squeeze victim during step 3, the index holds a stub over the deleted object, and a concurrent reader of E reads that stub.
A concurrent reader also hits the panic in the window between step 2 and the new value reaching the index.
Action Required
Remove the index entry for a MemorySqueezedLiquid entry in supersede_disk_copy, as the stub branch at line 490 already does. drop_memory_entry (line 593) releases the memory tally for that entry.
A squeezed entry reads back through its store object, so `supersede_disk_copy` cannot delete that object and leave the entry in the index: an insert that then fails with `CacheFull` left an entry whose next read panicked. The squeezed entry now goes the way of a disk stub. `drop_memory_entry` likewise discards the disk copy of the entry it drops, which the flush overflow path leaked as an orphaned object and a permanent reservation. The hydrated-liquid shortcut in `squeeze_victim_inner` is gone: float arrays squeeze without a hint, so it kept them from the squeezed tier. The policy always runs and `reuse_disk_copy` drops the rewrite.
| if let Some(previous) = previous { | ||
| // The put replaced the object under this key, so the previous | ||
| // copy's reservation goes with it. | ||
| self.budget.release_disk(previous.bytes); |
There was a problem hiding this comment.
nit: reserve the size delta instead of the full length when this put replaces an existing object (not blocking).
The reserve loop at line 976 asks for len while the previous copy still holds previous.bytes. Replacing an object therefore needs both sizes free at once, and the release at line 1014 happens only after the reserve succeeds. On a nearly full disk tier the reserve fails, so write_batch_to_disk returns CacheFull and the caller drops the entry — flush_all_to_disk calls drop_memory_entry at line 294 and line 324. The new flush_dropping_hydrated_entry_releases_its_disk_copy test shows that outcome: the entry disappears although the rewrite would not have grown the tier.
Read the record before the loop. Reserve len - previous.bytes when the value grows, and release previous.bytes - len after the put when it shrinks. Keep the record in place if the reserve fails, so discard_disk_copy still matches a live reservation.
There was a problem hiding this comment.
Agreed, and deferred to #46 rather than folded in here: reserving the delta changes what a flush does on a full tier, and flush_dropping_hydrated_entry_releases_its_disk_copy would need a different route to CacheFull (it relies on exactly this over-reservation today). The overwrite and drop paths in this PR keep the record and the reservation consistent either way.
There was a problem hiding this comment.
Prior blocking findings are addressed. supersede_disk_copy now routes a MemorySqueezedLiquid entry through drop_memory_entry, and drop_memory_entry discards the disk copy, so no index entry survives over a deleted store object. Removing the unhinted shortcut in squeeze_victim_inner lets a float entry reach the squeezed tier again, and reuse_disk_copy still drops the rewrite.
One non-blocking comment on write_batch_to_disk reservation ordering.
CI was still queued or in progress at review time, so the new tests are unverified here.
1 participant
Addresses #43 with the two defects a counting-allocator test reproduces locally in two seconds, independent of #44.
The index (
CongeeArc) frees a removed or replaced entry through crossbeam-epoch deferred destruction, so every evicted array stayed alive until a later collection: idle heap ran 6.14x the budget tally and the read-pass peak 6.09x the tier limit; the ART now holds a slot whose payload is taken out eagerly (1.01x and 1.71x after). Separately, a hydrated entry re-serialised and rewrote its own disk copy on eviction and reserved it twice (190 writes in a pure read pass, disk tally 2.7x the real objects);reuse_disk_copynow drops the write when the policy's chosen form is backed by the copy already on disk. The newmemory_footprinttest asserts all of it and fails onmain.Review follow-ups (7cb9f7c, 3ae1a4e, ad483ad): a caller overwrite drops the superseded disk copy, taking a stub or squeezed entry out of the index with it, so a demotion of the new value cannot flip to a stub over the old bytes and a failed insert cannot leave an entry over a deleted object; a flushed arrow entry is recorded as an Arrow copy; a put that replaces the object under a key releases the previous reservation;
drop_memory_entrydiscards the dropped entry's disk copy; andArtIndex::getretries a lookup that raced a replace. Each has a test that fails on the previous head. The threedatafusion-localsnapshots were refreshed for the removed rewrites.