kevy_store/
hash.rs

1//! `Store` hash commands.
2
3use crate::small_hash::{self, AddResult as HAddResult, SmallHashData};
4use crate::util::{parse_f64, parse_i64};
5use crate::value::{HashData, SmallBytes, Value, hash_field_weight};
6/// `(field, value)` pairs collected off either hash encoding.
7type FieldValuePairs = Vec<(Vec<u8>, Vec<u8>)>;
8
9use crate::{Entry, Store, StoreError, now_ns};
10use std::sync::Arc;
11
12impl Store {
13    // ---- hashes --------------------------------------------------------
14
15    /// Borrow the key's hash mutably, optionally creating it. `Ok(None)` means
16    /// the key is absent and `create` was false.
17    ///
18    /// A.8: only used by the heap-backed legacy read/mutate sites
19    /// (`hincrby`, the `hash_ref` reader). The bulk writers (`hset` /
20    /// `hdel`) take the encoding-switch path via `hset_one` /
21    /// `hash_field_get_mut_for_hset`. When `create=true` on a missing
22    /// key, materialises a heap `Value::Hash(Arc::default())` (no inline
23    /// path), matching pre-A.8 behaviour for read-modify-write entry
24    /// points that don't carry per-pair size info.
25    fn hash_mut(&mut self, key: &[u8], create: bool) -> Result<Option<&mut HashData>, StoreError> {
26        if self.live_entry_mut(key).is_none() {
27            if !create {
28                return Ok(None);
29            }
30            self.insert_entry(
31                SmallBytes::from_slice(key),
32                Entry::new(Value::Hash(Arc::default()), None),
33            );
34        }
35        // A.8: detect inline-encoding first (independent borrow), then
36        // upgrade out-of-scope of the &mut, then re-borrow as Hash. The
37        // borrow checker rejects the obvious in-place match because
38        // both arms would return a borrow tied to the same `&mut self`.
39        let is_inline = matches!(
40            self.map.get(key).map(|e| &e.value),
41            Some(Value::SmallHashInline(_))
42        );
43        if is_inline {
44            let promoted = {
45                let e = self.map.get(key).expect("present");
46                if let Value::SmallHashInline(s) = &e.value {
47                    small_hash::promote(s)
48                } else {
49                    unreachable!()
50                }
51            };
52            self.map.get_mut(key).expect("present").value = Value::Hash(Arc::new(promoted));
53            self.reweigh_entry(key);
54        }
55        match &mut self.map.get_mut(key).expect("present").value {
56            Value::Hash(h) => Ok(Some(Arc::make_mut(h))),
57            _ => Err(StoreError::WrongType),
58        }
59    }
60
61    /// A.8: read the key's hash slot for HSET. `WrongType` if the entry
62    /// is not a hash. Returns `None` when the key is absent — caller
63    /// (`hset_one`) creates the entry then.
64    fn hash_value_for_set(&mut self, key: &[u8]) -> Result<Option<&mut Value>, StoreError> {
65        match self.live_entry_mut(key) {
66            None => Ok(None),
67            Some(e) => match &e.value {
68                Value::Hash(_) | Value::SmallHashInline(_) => Ok(Some(&mut e.value)),
69                _ => Err(StoreError::WrongType),
70            },
71        }
72    }
73
74    /// Read the key's hash immutably (lazily expiring) — returns the
75    /// pairs as a vector of `(&[u8], &[u8])`. None if absent.
76    /// Internal helper for read-only paths; collects into a new Vec to
77    /// avoid the two-encoding match dance at every callsite.
78    fn hash_pairs(&mut self, key: &[u8]) -> Result<Option<FieldValuePairs>, StoreError> {
79        match self.live_entry(key) {
80            None => Ok(None),
81            Some(e) => match &e.value {
82                Value::Hash(h) => Ok(Some(
83                    h.iter().map(|(f, v)| (f.to_vec(), v.clone())).collect(),
84                )),
85                Value::SmallHashInline(h) => Ok(Some(
86                    h.iter().map(|(f, v)| (f.to_vec(), v.to_vec())).collect(),
87                )),
88                _ => Err(StoreError::WrongType),
89            },
90        }
91    }
92
93    /// G4 (v1.25): borrowed-pair `HSET` — kills the per-field+value
94    /// `Vec<u8>` allocs the dispatch layer used to do before calling
95    /// [`Self::hset`]. A.8: routes through the encoding-switch path.
96    pub fn hset_borrowed(
97        &mut self,
98        key: &[u8],
99        pairs: &[(&[u8], &[u8])],
100    ) -> Result<usize, StoreError> {
101        self.purge_hash_ttl(key);
102        // v2.4: overwriting a field discards its TTL (Redis 7.4).
103        if !self.hfttl.is_empty() {
104            let fs: Vec<&[u8]> = pairs.iter().map(|(f, _)| *f).collect();
105            self.clear_hash_field_ttls(key, &fs);
106        }
107        if pairs.is_empty() {
108            return Ok(0);
109        }
110        let mut added = 0usize;
111        let mut delta: i64 = 0;
112        for (f, v) in pairs {
113            match self.hset_one(key, f, v)? {
114                HsetOutcome::AddedInline => {
115                    added += 1;
116                    // Inline carries zero heap delta — already accounted
117                    // at insert_entry / per-call via value.weight()==0.
118                }
119                HsetOutcome::UpdatedInline => {}
120                HsetOutcome::AddedHeap(w) => {
121                    added += 1;
122                    delta += w;
123                }
124                HsetOutcome::UpdatedHeap(d) => {
125                    delta += d;
126                }
127            }
128        }
129        self.account_delta(key, delta);
130        Ok(added)
131    }
132
133    /// `HSET` — returns the count of newly-added fields.
134    pub fn hset(&mut self, key: &[u8], pairs: &[(Vec<u8>, Vec<u8>)]) -> Result<usize, StoreError> {
135        self.purge_hash_ttl(key);
136        if !self.hfttl.is_empty() {
137            let fs: Vec<&[u8]> = pairs.iter().map(|(f, _)| f.as_slice()).collect();
138            self.clear_hash_field_ttls(key, &fs);
139        }
140        let borrowed: Vec<(&[u8], &[u8])> =
141            pairs.iter().map(|(f, v)| (f.as_slice(), v.as_slice())).collect();
142        self.hset_borrowed(key, &borrowed)
143    }
144
145    /// `HSETNX` — set only if the field is absent; returns whether it was set.
146    pub fn hsetnx(&mut self, key: &[u8], field: &[u8], val: &[u8]) -> Result<bool, StoreError> {
147        self.purge_hash_ttl(key);
148        // Existing-field fast check via the encoding-aware reader.
149        let exists = match self.live_entry(key) {
150            None => false,
151            Some(e) => match &e.value {
152                Value::Hash(h) => h.contains_key(field),
153                Value::SmallHashInline(h) => h.contains_key(field),
154                _ => return Err(StoreError::WrongType),
155            },
156        };
157        if exists {
158            return Ok(false);
159        }
160        match self.hset_one(key, field, val)? {
161            HsetOutcome::AddedInline | HsetOutcome::UpdatedInline => Ok(true),
162            HsetOutcome::AddedHeap(w) => {
163                self.account_delta(key, w);
164                Ok(true)
165            }
166            HsetOutcome::UpdatedHeap(_) => Ok(true),
167        }
168    }
169
170    pub fn hget(&mut self, key: &[u8], field: &[u8]) -> Result<Option<&[u8]>, StoreError> {
171        self.purge_hash_ttl(key);
172        match self.live_entry(key) {
173            None => Ok(None),
174            Some(e) => match &e.value {
175                Value::Hash(h) => Ok(h.get(field).map(Vec::as_slice)),
176                Value::SmallHashInline(h) => Ok(h.get(field)),
177                _ => Err(StoreError::WrongType),
178            },
179        }
180    }
181
182    pub fn hexists(&mut self, key: &[u8], field: &[u8]) -> Result<bool, StoreError> {
183        self.purge_hash_ttl(key);
184        match self.live_entry(key) {
185            None => Ok(false),
186            Some(e) => match &e.value {
187                Value::Hash(h) => Ok(h.contains_key(field)),
188                Value::SmallHashInline(h) => Ok(h.contains_key(field)),
189                _ => Err(StoreError::WrongType),
190            },
191        }
192    }
193
194    pub fn hlen(&mut self, key: &[u8]) -> Result<usize, StoreError> {
195        self.purge_hash_ttl(key);
196        match self.live_entry(key) {
197            None => Ok(0),
198            Some(e) => match &e.value {
199                Value::Hash(h) => Ok(h.len()),
200                Value::SmallHashInline(h) => Ok(h.len()),
201                _ => Err(StoreError::WrongType),
202            },
203        }
204    }
205
206    pub fn hmget(
207        &mut self,
208        key: &[u8],
209        fields: &[Vec<u8>],
210    ) -> Result<Vec<Option<Vec<u8>>>, StoreError> {
211        self.purge_hash_ttl(key);
212        let borrowed: Vec<&[u8]> = fields.iter().map(Vec::as_slice).collect();
213        self.hmget_borrowed(key, &borrowed)
214    }
215
216    /// G4 (v1.25): borrowed-slice `HMGET`.
217    pub fn hmget_borrowed(
218        &mut self,
219        key: &[u8],
220        fields: &[&[u8]],
221    ) -> Result<Vec<Option<Vec<u8>>>, StoreError> {
222        self.purge_hash_ttl(key);
223        match self.live_entry(key) {
224            None => Ok(fields.iter().map(|_| None).collect()),
225            Some(e) => match &e.value {
226                Value::Hash(h) => Ok(fields.iter().map(|f| h.get(*f).cloned()).collect()),
227                Value::SmallHashInline(h) => Ok(fields
228                    .iter()
229                    .map(|f| h.get(f).map(<[u8]>::to_vec))
230                    .collect()),
231                _ => Err(StoreError::WrongType),
232            },
233        }
234    }
235
236    /// `HGETALL` — flat `[field, value, field, value, ...]`.
237    pub fn hgetall(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>, StoreError> {
238        self.purge_hash_ttl(key);
239        match self.hash_pairs(key)? {
240            None => Ok(Vec::new()),
241            Some(pairs) => {
242                let mut out = Vec::with_capacity(pairs.len() * 2);
243                for (f, v) in pairs {
244                    out.push(f);
245                    out.push(v);
246                }
247                Ok(out)
248            }
249        }
250    }
251
252    pub fn hkeys(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>, StoreError> {
253        self.purge_hash_ttl(key);
254        match self.live_entry(key) {
255            None => Ok(Vec::new()),
256            Some(e) => match &e.value {
257                Value::Hash(h) => Ok(h.keys().map(kevy_bytes::SmallBytes::to_vec).collect()),
258                Value::SmallHashInline(h) => Ok(h.iter().map(|(f, _)| f.to_vec()).collect()),
259                _ => Err(StoreError::WrongType),
260            },
261        }
262    }
263
264    pub fn hvals(&mut self, key: &[u8]) -> Result<Vec<Vec<u8>>, StoreError> {
265        self.purge_hash_ttl(key);
266        match self.live_entry(key) {
267            None => Ok(Vec::new()),
268            Some(e) => match &e.value {
269                Value::Hash(h) => Ok(h.values().cloned().collect()),
270                Value::SmallHashInline(h) => Ok(h.iter().map(|(_, v)| v.to_vec()).collect()),
271                _ => Err(StoreError::WrongType),
272            },
273        }
274    }
275
276    /// `HDEL` — returns count removed; deletes the key if hash becomes empty.
277    pub fn hdel(&mut self, key: &[u8], fields: &[Vec<u8>]) -> Result<usize, StoreError> {
278        self.purge_hash_ttl(key);
279        let borrowed: Vec<&[u8]> = fields.iter().map(Vec::as_slice).collect();
280        self.hdel_borrowed(key, &borrowed)
281    }
282
283    /// G4 (v1.25): borrowed-slice `HDEL`. A.8: encoding-aware.
284    pub fn hdel_borrowed(
285        &mut self,
286        key: &[u8],
287        fields: &[&[u8]],
288    ) -> Result<usize, StoreError> {
289        self.purge_hash_ttl(key);
290        let now = now_ns();
291        if !self.reap(key, now) {
292            return Ok(0);
293        }
294        let (removed, delta, drop_key) = {
295            let h_entry = self.map.get_mut(key).expect("live");
296            match &mut h_entry.value {
297                Value::Hash(h) => {
298                    // G-A3: hoist Arc::make_mut OUT of the loop — done
299                    // once per command instead of per-field.
300                    let h = Arc::make_mut(h);
301                    let mut r = 0usize;
302                    let mut d: i64 = 0;
303                    for f in fields {
304                        if let Some(old_v) = h.remove(*f) {
305                            r += 1;
306                            let smb = SmallBytes::from_slice(f);
307                            d -= hash_field_weight(&smb, old_v.len()) as i64;
308                        }
309                    }
310                    let drop_now = h.is_empty();
311                    (r, d, drop_now)
312                }
313                Value::SmallHashInline(h) => {
314                    let mut r = 0usize;
315                    for f in fields {
316                        if h.try_remove(f) {
317                            r += 1;
318                        }
319                    }
320                    let drop_now = h.is_empty();
321                    (r, 0i64, drop_now)
322                }
323                _ => return Err(StoreError::WrongType),
324            }
325        };
326        if drop_key {
327            self.remove_entry(key);
328        } else {
329            self.account_delta(key, delta);
330        }
331        Ok(removed)
332    }
333
334    /// `HINCRBYFLOAT` — atomic float increment of a hash field.
335    /// Preserves TTL; errors with `NotFloat` if the field isn't a
336    /// parseable float. Returns the post-increment value.
337    pub fn hincrbyfloat(
338        &mut self,
339        key: &[u8],
340        field: &[u8],
341        delta: f64,
342    ) -> Result<f64, StoreError> {
343        self.purge_hash_ttl(key);
344        self.clear_hash_field_ttls(key, &[field]);
345        let (next, weight_delta) = {
346            let h = self.hash_mut(key, true)?.expect("created");
347            let cur = match h.get(field) {
348                Some(v) => parse_f64(v).ok_or(StoreError::NotFloat)?,
349                None => 0.0,
350            };
351            let next = cur + delta;
352            if !next.is_finite() {
353                return Err(StoreError::NotFloat);
354            }
355            let new_bytes = format!("{next}").into_bytes();
356            let smb = SmallBytes::from_slice(field);
357            let new_field_w = hash_field_weight(&smb, new_bytes.len()) as i64;
358            let new_value_len = new_bytes.len();
359            let wd = match h.insert(smb, new_bytes) {
360                None => new_field_w,
361                Some(old) => new_value_len as i64 - old.len() as i64,
362            };
363            (next, wd)
364        };
365        self.account_delta(key, weight_delta);
366        Ok(next)
367    }
368
369    /// `HINCRBY` — preserves TTL; errors if the field isn't an integer.
370    pub fn hincrby(&mut self, key: &[u8], field: &[u8], delta: i64) -> Result<i64, StoreError> {
371        self.purge_hash_ttl(key);
372        self.clear_hash_field_ttls(key, &[field]);
373        let (next, weight_delta) = {
374            let h = self.hash_mut(key, true)?.expect("created");
375            let cur = match h.get(field) {
376                Some(v) => parse_i64(v).ok_or(StoreError::NotInteger)?,
377                None => 0,
378            };
379            let next = cur.checked_add(delta).ok_or(StoreError::Overflow)?;
380            let new_bytes = next.to_string().into_bytes();
381            let smb = SmallBytes::from_slice(field);
382            let new_field_w = hash_field_weight(&smb, new_bytes.len()) as i64;
383            let new_value_len = new_bytes.len();
384            let wd = match h.insert(smb, new_bytes) {
385                None => new_field_w,
386                Some(old) => new_value_len as i64 - old.len() as i64,
387            };
388            (next, wd)
389        };
390        self.account_delta(key, weight_delta);
391        Ok(next)
392    }
393
394    /// A.8 core: set one `(field, value)` pair, applying the
395    /// encoding-switch. Returns the per-call outcome (whether added or
396    /// updated, and whether it sits in the inline or heap variant).
397    fn hset_one(
398        &mut self,
399        key: &[u8],
400        field: &[u8],
401        value: &[u8],
402    ) -> Result<HsetOutcome, StoreError> {
403        // Missing key — pick encoding by first pair size.
404        if self.hash_value_for_set(key)?.is_none() {
405            return Ok(self.hset_create(key, field, value));
406        }
407        let v = self.hash_value_for_set(key)?.expect("present and a hash");
408        match v {
409            Value::SmallHashInline(h) => match h.try_set(field, value) {
410                HAddResult::Added => Ok(HsetOutcome::AddedInline),
411                HAddResult::Updated => Ok(HsetOutcome::UpdatedInline),
412                HAddResult::NoRoom => {
413                    // Promote inline → Hash(Arc<HashData>), then set
414                    // (handles the spilling pair).
415                    let mut promoted = small_hash::promote(h);
416                    let smb = SmallBytes::from_slice(field);
417                    let new_w = hash_field_weight(&smb, value.len()) as i64;
418                    let added = !promoted.contains_key(field);
419                    let prior_v_len = promoted.get(field).map_or(0, Vec::len);
420                    promoted.insert(smb, value.to_vec());
421                    *v = Value::Hash(Arc::new(promoted));
422                    self.reweigh_entry(key);
423                    if added {
424                        Ok(HsetOutcome::AddedHeap(new_w))
425                    } else {
426                        Ok(HsetOutcome::UpdatedHeap(value.len() as i64 - prior_v_len as i64))
427                    }
428                }
429            },
430            Value::Hash(h) => {
431                let h = Arc::make_mut(h);
432                let smb = SmallBytes::from_slice(field);
433                let new_w = hash_field_weight(&smb, value.len()) as i64;
434                let new_value_len = value.len();
435                match h.insert(smb, value.to_vec()) {
436                    None => Ok(HsetOutcome::AddedHeap(new_w)),
437                    Some(old) => {
438                        Ok(HsetOutcome::UpdatedHeap(new_value_len as i64 - old.len() as i64))
439                    }
440                }
441            }
442            _ => Err(StoreError::WrongType),
443        }
444    }
445
446    /// Create a fresh entry for `key` holding one pair. Picks inline
447    /// when both field + value fit, falls back to heap otherwise.
448    fn hset_create(&mut self, key: &[u8], field: &[u8], value: &[u8]) -> HsetOutcome {
449        if let Some(inline) = SmallHashData::with_one(field, value) {
450            self.insert_entry(
451                SmallBytes::from_slice(key),
452                Entry::new(Value::SmallHashInline(inline), None),
453            );
454            // Insert already accounts via value.weight() == 0; per-pair
455            // delta is zero in the caller (matches inline arm).
456            HsetOutcome::AddedInline
457        } else {
458            let smb_f = SmallBytes::from_slice(field);
459            let mut h = HashData::with_capacity(1);
460            h.insert(smb_f, value.to_vec());
461            self.insert_entry(
462                SmallBytes::from_slice(key),
463                Entry::new(Value::Hash(Arc::new(h)), None),
464            );
465            HsetOutcome::AddedInline
466        }
467    }
468
469}
470
471enum HsetOutcome {
472    /// Field was new and lives in the inline variant (zero heap delta).
473    AddedInline,
474    /// Field existed in the inline variant (no count bump, no delta).
475    UpdatedInline,
476    /// Field was new in the heap variant; carries the new field's weight.
477    AddedHeap(i64),
478    /// Field existed in the heap variant; carries the value-length delta.
479    UpdatedHeap(i64),
480}
kevy_store/hash.rs

kevy_store/
hash.rs
