hyphae_storage/
mutation.rs

1// SPDX-License-Identifier: Apache-2.0
2
3use hyphae_core::{
4    Q15Vector, VectorMetric, VectorSpaceDefinition, VectorSpaceName, VectorValueError,
5};
6use hyphae_query::FieldPath;
7use hyphae_retrieval::{LexicalField, LexicalIndexDefinition};
8use thiserror::Error;
9
10use crate::log::MAX_OPERATION_BYTES;
11
12const PUT: u8 = 1;
13const DELETE: u8 = 2;
14const DEFINE_VECTOR_SPACE: u8 = 3;
15const UPSERT_VECTOR: u8 = 4;
16const DELETE_VECTOR: u8 = 5;
17const DEFINE_LEXICAL_INDEX: u8 = 6;
18const PUT_HEADER_LENGTH: usize = 13;
19const DELETE_HEADER_LENGTH: usize = 5;
20
21/// Maximum encoded key length accepted by the embedded KV layer.
22pub const MAX_KEY_BYTES: usize = 1024 * 1024;
23
24/// A deterministic mutation persisted inside one transaction operation frame.
25#[derive(Clone, Debug, Eq, PartialEq)]
26pub enum Mutation {
27    /// Replaces the value stored under a nonempty binary key.
28    Put {
29        /// Binary key.
30        key: Vec<u8>,
31        /// Binary value.
32        value: Vec<u8>,
33    },
34    /// Removes a binary key. Deleting a missing key is idempotent.
35    Delete {
36        /// Binary key.
37        key: Vec<u8>,
38    },
39    /// Defines one immutable named vector space.
40    DefineVectorSpace {
41        /// Canonical vector-space definition.
42        definition: VectorSpaceDefinition,
43    },
44    /// Replaces one vector at `(space, key)`.
45    UpsertVector {
46        /// Canonical vector-space identifier.
47        space: VectorSpaceName,
48        /// Binary object key.
49        key: Vec<u8>,
50        /// Canonical Q15 vector.
51        vector: Q15Vector,
52    },
53    /// Removes one vector. Deleting a missing vector is idempotent.
54    DeleteVector {
55        /// Canonical vector-space identifier.
56        space: VectorSpaceName,
57        /// Binary object key.
58        key: Vec<u8>,
59    },
60    /// Defines one immutable named lexical index.
61    DefineLexicalIndex {
62        /// Canonical lexical definition.
63        definition: LexicalIndexDefinition,
64    },
65}
66
67impl Mutation {
68    /// Creates a put mutation.
69    pub fn put(key: impl Into<Vec<u8>>, value: impl Into<Vec<u8>>) -> Self {
70        Self::Put {
71            key: key.into(),
72            value: value.into(),
73        }
74    }
75
76    /// Creates a delete mutation.
77    pub fn delete(key: impl Into<Vec<u8>>) -> Self {
78        Self::Delete { key: key.into() }
79    }
80
81    /// Creates a vector-space definition mutation.
82    pub fn define_vector_space(definition: VectorSpaceDefinition) -> Self {
83        Self::DefineVectorSpace { definition }
84    }
85
86    /// Creates a durable vector upsert mutation.
87    pub fn upsert_vector(
88        space: VectorSpaceName,
89        key: impl Into<Vec<u8>>,
90        vector: Q15Vector,
91    ) -> Self {
92        Self::UpsertVector {
93            space,
94            key: key.into(),
95            vector,
96        }
97    }
98
99    /// Creates a durable vector delete mutation.
100    pub fn delete_vector(space: VectorSpaceName, key: impl Into<Vec<u8>>) -> Self {
101        Self::DeleteVector {
102            space,
103            key: key.into(),
104        }
105    }
106
107    /// Creates a lexical-index definition mutation.
108    pub fn define_lexical_index(definition: LexicalIndexDefinition) -> Self {
109        Self::DefineLexicalIndex { definition }
110    }
111
112    pub(crate) fn encode(&self) -> Result<Vec<u8>, MutationError> {
113        match self {
114            Self::Put { key, value } => encode_put(key, value),
115            Self::Delete { key } => encode_delete(key),
116            Self::DefineVectorSpace { definition } => encode_vector_space(definition),
117            Self::UpsertVector { space, key, vector } => encode_vector(space, key, vector),
118            Self::DeleteVector { space, key } => encode_vector_delete(space, key),
119            Self::DefineLexicalIndex { definition } => encode_lexical_index(definition),
120        }
121    }
122
123    pub(crate) fn decode(encoded: &[u8]) -> Result<Self, MutationError> {
124        let Some(kind) = encoded.first().copied() else {
125            return Err(MutationError::Malformed);
126        };
127        match kind {
128            PUT => decode_put(encoded),
129            DELETE => decode_delete(encoded),
130            DEFINE_VECTOR_SPACE => decode_vector_space(encoded),
131            UPSERT_VECTOR => decode_vector(encoded),
132            DELETE_VECTOR => decode_vector_delete(encoded),
133            DEFINE_LEXICAL_INDEX => decode_lexical_index(encoded),
134            kind => Err(MutationError::UnknownKind { kind }),
135        }
136    }
137}
138
139/// Failure while validating or decoding a persisted mutation.
140#[derive(Clone, Debug, Error, Eq, PartialEq)]
141pub enum MutationError {
142    /// Empty keys are forbidden.
143    #[error("mutation key must not be empty")]
144    EmptyKey,
145
146    /// A key exceeds the stable storage limit.
147    #[error("mutation key is {length} bytes; maximum is {maximum}")]
148    KeyTooLarge {
149        /// Actual key length.
150        length: usize,
151        /// Maximum key length.
152        maximum: usize,
153    },
154
155    /// The full encoded operation exceeds one log frame.
156    #[error("encoded mutation is {length} bytes; maximum is {maximum}")]
157    OperationTooLarge {
158        /// Actual encoded length.
159        length: usize,
160        /// Maximum operation length.
161        maximum: usize,
162    },
163
164    /// Persisted bytes do not form one canonical mutation.
165    #[error("malformed persisted mutation")]
166    Malformed,
167
168    /// The mutation kind is not defined by this disk format.
169    #[error("unknown persisted mutation kind {kind}")]
170    UnknownKind {
171        /// Raw kind byte.
172        kind: u8,
173    },
174
175    /// A canonical shared vector value is invalid.
176    #[error(transparent)]
177    Vector(#[from] VectorValueError),
178
179    /// A canonical lexical definition is invalid.
180    #[error("invalid lexical definition")]
181    Lexical,
182}
183
184fn encode_lexical_index(definition: &LexicalIndexDefinition) -> Result<Vec<u8>, MutationError> {
185    let mut encoded = Vec::new();
186    encoded.push(DEFINE_LEXICAL_INDEX);
187    encode_space(&definition.name, &mut encoded)?;
188    let field_count = u8::try_from(definition.fields.len()).map_err(|_| MutationError::Lexical)?;
189    encoded.push(field_count);
190    for field in &definition.fields {
191        let segment_count =
192            u8::try_from(field.path.segments().len()).map_err(|_| MutationError::Lexical)?;
193        encoded.push(segment_count);
194        for segment in field.path.segments() {
195            let bytes = segment.as_bytes();
196            let length = u16::try_from(bytes.len()).map_err(|_| MutationError::Lexical)?;
197            encoded.extend_from_slice(&length.to_le_bytes());
198            encoded.extend_from_slice(bytes);
199        }
200        encoded.extend_from_slice(&field.weight_micros.to_le_bytes());
201    }
202    validate_operation_length(encoded.len())?;
203    Ok(encoded)
204}
205
206fn decode_lexical_index(encoded: &[u8]) -> Result<Mutation, MutationError> {
207    let mut cursor = 1;
208    let name = decode_space(encoded, &mut cursor)?;
209    let field_count = usize::from(*encoded.get(cursor).ok_or(MutationError::Malformed)?);
210    cursor = cursor.checked_add(1).ok_or(MutationError::Malformed)?;
211    let mut fields = Vec::with_capacity(field_count);
212    for _ in 0..field_count {
213        let segment_count = usize::from(*encoded.get(cursor).ok_or(MutationError::Malformed)?);
214        cursor = cursor.checked_add(1).ok_or(MutationError::Malformed)?;
215        let mut segments = Vec::with_capacity(segment_count);
216        for _ in 0..segment_count {
217            let end = cursor.checked_add(2).ok_or(MutationError::Malformed)?;
218            let length = usize::from(u16::from_le_bytes(copy_array(
219                encoded.get(cursor..end).ok_or(MutationError::Malformed)?,
220            )));
221            cursor = end;
222            let end = cursor.checked_add(length).ok_or(MutationError::Malformed)?;
223            let segment =
224                std::str::from_utf8(encoded.get(cursor..end).ok_or(MutationError::Malformed)?)
225                    .map_err(|_| MutationError::Malformed)?
226                    .to_owned();
227            cursor = end;
228            segments.push(segment);
229        }
230        let end = cursor.checked_add(4).ok_or(MutationError::Malformed)?;
231        let weight_micros = u32::from_le_bytes(copy_array(
232            encoded.get(cursor..end).ok_or(MutationError::Malformed)?,
233        ));
234        cursor = end;
235        fields.push(LexicalField {
236            path: FieldPath::new(segments),
237            weight_micros,
238        });
239    }
240    if cursor != encoded.len() {
241        return Err(MutationError::Malformed);
242    }
243    let definition =
244        LexicalIndexDefinition::new(name, fields).map_err(|_| MutationError::Lexical)?;
245    Ok(Mutation::DefineLexicalIndex { definition })
246}
247
248fn encode_space(space: &VectorSpaceName, encoded: &mut Vec<u8>) -> Result<(), MutationError> {
249    let length = u8::try_from(space.as_str().len()).map_err(|_| MutationError::Malformed)?;
250    encoded.push(length);
251    encoded.extend_from_slice(space.as_str().as_bytes());
252    Ok(())
253}
254
255fn decode_space(encoded: &[u8], cursor: &mut usize) -> Result<VectorSpaceName, MutationError> {
256    let length = usize::from(*encoded.get(*cursor).ok_or(MutationError::Malformed)?);
257    *cursor = cursor.checked_add(1).ok_or(MutationError::Malformed)?;
258    let end = cursor.checked_add(length).ok_or(MutationError::Malformed)?;
259    let raw = encoded.get(*cursor..end).ok_or(MutationError::Malformed)?;
260    *cursor = end;
261    let value = std::str::from_utf8(raw).map_err(|_| MutationError::Malformed)?;
262    Ok(VectorSpaceName::new(value.to_owned())?)
263}
264
265fn encode_vector_space(definition: &VectorSpaceDefinition) -> Result<Vec<u8>, MutationError> {
266    let mut encoded = Vec::with_capacity(1 + 1 + definition.name.as_str().len() + 4);
267    encoded.push(DEFINE_VECTOR_SPACE);
268    encode_space(&definition.name, &mut encoded)?;
269    encoded.extend_from_slice(&definition.dimension.to_le_bytes());
270    encoded.push(definition.metric as u8);
271    encoded.push(1);
272    validate_operation_length(encoded.len())?;
273    Ok(encoded)
274}
275
276fn decode_vector_space(encoded: &[u8]) -> Result<Mutation, MutationError> {
277    let mut cursor = 1;
278    let name = decode_space(encoded, &mut cursor)?;
279    let dimension_end = cursor.checked_add(2).ok_or(MutationError::Malformed)?;
280    let dimension = u16::from_le_bytes(copy_array(
281        encoded
282            .get(cursor..dimension_end)
283            .ok_or(MutationError::Malformed)?,
284    ));
285    cursor = dimension_end;
286    if encoded.get(cursor) != Some(&(VectorMetric::Cosine as u8))
287        || encoded.get(cursor + 1) != Some(&1)
288        || cursor + 2 != encoded.len()
289    {
290        return Err(MutationError::Malformed);
291    }
292    let definition = VectorSpaceDefinition::cosine(name, dimension)?;
293    Ok(Mutation::DefineVectorSpace { definition })
294}
295
296fn encode_vector(
297    space: &VectorSpaceName,
298    key: &[u8],
299    vector: &Q15Vector,
300) -> Result<Vec<u8>, MutationError> {
301    validate_key(key)?;
302    let key_length = u32::try_from(key.len()).map_err(|_| MutationError::KeyTooLarge {
303        length: key.len(),
304        maximum: MAX_KEY_BYTES,
305    })?;
306    let mut encoded = Vec::with_capacity(
307        1 + 1 + space.as_str().len() + 4 + key.len() + 2 + 2 * vector.as_slice().len(),
308    );
309    encoded.push(UPSERT_VECTOR);
310    encode_space(space, &mut encoded)?;
311    encoded.extend_from_slice(&key_length.to_le_bytes());
312    encoded.extend_from_slice(key);
313    encoded.extend_from_slice(&vector.dimension().to_le_bytes());
314    for value in vector.as_slice() {
315        encoded.extend_from_slice(&value.to_le_bytes());
316    }
317    validate_operation_length(encoded.len())?;
318    Ok(encoded)
319}
320
321fn decode_vector(encoded: &[u8]) -> Result<Mutation, MutationError> {
322    let mut cursor = 1;
323    let space = decode_space(encoded, &mut cursor)?;
324    let key_length_end = cursor.checked_add(4).ok_or(MutationError::Malformed)?;
325    let key_length = usize::try_from(u32::from_le_bytes(copy_array(
326        encoded
327            .get(cursor..key_length_end)
328            .ok_or(MutationError::Malformed)?,
329    )))
330    .map_err(|_| MutationError::Malformed)?;
331    cursor = key_length_end;
332    let key_end = cursor
333        .checked_add(key_length)
334        .ok_or(MutationError::Malformed)?;
335    let key = encoded
336        .get(cursor..key_end)
337        .ok_or(MutationError::Malformed)?
338        .to_vec();
339    validate_key(&key)?;
340    cursor = key_end;
341    let dimension_end = cursor.checked_add(2).ok_or(MutationError::Malformed)?;
342    let dimension = usize::from(u16::from_le_bytes(copy_array(
343        encoded
344            .get(cursor..dimension_end)
345            .ok_or(MutationError::Malformed)?,
346    )));
347    cursor = dimension_end;
348    let vector_bytes = dimension.checked_mul(2).ok_or(MutationError::Malformed)?;
349    let vector_end = cursor
350        .checked_add(vector_bytes)
351        .ok_or(MutationError::Malformed)?;
352    if vector_end != encoded.len() {
353        return Err(MutationError::Malformed);
354    }
355    let mut values = Vec::with_capacity(dimension);
356    for chunk in encoded[cursor..vector_end].chunks_exact(2) {
357        values.push(i16::from_le_bytes(copy_array(chunk)));
358    }
359    Ok(Mutation::UpsertVector {
360        space,
361        key,
362        vector: Q15Vector::new(values)?,
363    })
364}
365
366fn encode_vector_delete(space: &VectorSpaceName, key: &[u8]) -> Result<Vec<u8>, MutationError> {
367    validate_key(key)?;
368    let key_length = u32::try_from(key.len()).map_err(|_| MutationError::KeyTooLarge {
369        length: key.len(),
370        maximum: MAX_KEY_BYTES,
371    })?;
372    let mut encoded = Vec::with_capacity(1 + 1 + space.as_str().len() + 4 + key.len());
373    encoded.push(DELETE_VECTOR);
374    encode_space(space, &mut encoded)?;
375    encoded.extend_from_slice(&key_length.to_le_bytes());
376    encoded.extend_from_slice(key);
377    validate_operation_length(encoded.len())?;
378    Ok(encoded)
379}
380
381fn decode_vector_delete(encoded: &[u8]) -> Result<Mutation, MutationError> {
382    let mut cursor = 1;
383    let space = decode_space(encoded, &mut cursor)?;
384    let key_length_end = cursor.checked_add(4).ok_or(MutationError::Malformed)?;
385    let key_length = usize::try_from(u32::from_le_bytes(copy_array(
386        encoded
387            .get(cursor..key_length_end)
388            .ok_or(MutationError::Malformed)?,
389    )))
390    .map_err(|_| MutationError::Malformed)?;
391    cursor = key_length_end;
392    let key_end = cursor
393        .checked_add(key_length)
394        .ok_or(MutationError::Malformed)?;
395    if key_end != encoded.len() {
396        return Err(MutationError::Malformed);
397    }
398    let key = encoded[cursor..key_end].to_vec();
399    validate_key(&key)?;
400    Ok(Mutation::DeleteVector { space, key })
401}
402
403fn encode_put(key: &[u8], value: &[u8]) -> Result<Vec<u8>, MutationError> {
404    validate_key(key)?;
405    let key_length = u32::try_from(key.len()).map_err(|_| MutationError::KeyTooLarge {
406        length: key.len(),
407        maximum: MAX_KEY_BYTES,
408    })?;
409    let value_length =
410        u64::try_from(value.len()).map_err(|_| MutationError::OperationTooLarge {
411            length: usize::MAX,
412            maximum: MAX_OPERATION_BYTES,
413        })?;
414    let encoded_length = PUT_HEADER_LENGTH
415        .checked_add(key.len())
416        .and_then(|length| length.checked_add(value.len()))
417        .ok_or(MutationError::OperationTooLarge {
418            length: usize::MAX,
419            maximum: MAX_OPERATION_BYTES,
420        })?;
421    validate_operation_length(encoded_length)?;
422
423    let mut encoded = Vec::with_capacity(encoded_length);
424    encoded.push(PUT);
425    encoded.extend_from_slice(&key_length.to_le_bytes());
426    encoded.extend_from_slice(&value_length.to_le_bytes());
427    encoded.extend_from_slice(key);
428    encoded.extend_from_slice(value);
429    Ok(encoded)
430}
431
432fn encode_delete(key: &[u8]) -> Result<Vec<u8>, MutationError> {
433    validate_key(key)?;
434    let key_length = u32::try_from(key.len()).map_err(|_| MutationError::KeyTooLarge {
435        length: key.len(),
436        maximum: MAX_KEY_BYTES,
437    })?;
438    let encoded_length =
439        DELETE_HEADER_LENGTH
440            .checked_add(key.len())
441            .ok_or(MutationError::OperationTooLarge {
442                length: usize::MAX,
443                maximum: MAX_OPERATION_BYTES,
444            })?;
445    validate_operation_length(encoded_length)?;
446
447    let mut encoded = Vec::with_capacity(encoded_length);
448    encoded.push(DELETE);
449    encoded.extend_from_slice(&key_length.to_le_bytes());
450    encoded.extend_from_slice(key);
451    Ok(encoded)
452}
453
454fn decode_put(encoded: &[u8]) -> Result<Mutation, MutationError> {
455    if encoded.len() < PUT_HEADER_LENGTH {
456        return Err(MutationError::Malformed);
457    }
458    let key_length = usize::try_from(u32::from_le_bytes(copy_array(&encoded[1..5])))
459        .map_err(|_| MutationError::Malformed)?;
460    let value_length = usize::try_from(u64::from_le_bytes(copy_array(&encoded[5..13])))
461        .map_err(|_| MutationError::Malformed)?;
462    let key_end = PUT_HEADER_LENGTH
463        .checked_add(key_length)
464        .ok_or(MutationError::Malformed)?;
465    let value_end = key_end
466        .checked_add(value_length)
467        .ok_or(MutationError::Malformed)?;
468    if value_end != encoded.len() {
469        return Err(MutationError::Malformed);
470    }
471    let key = encoded[PUT_HEADER_LENGTH..key_end].to_vec();
472    validate_key(&key)?;
473    validate_operation_length(encoded.len())?;
474    Ok(Mutation::Put {
475        key,
476        value: encoded[key_end..value_end].to_vec(),
477    })
478}
479
480fn decode_delete(encoded: &[u8]) -> Result<Mutation, MutationError> {
481    if encoded.len() < DELETE_HEADER_LENGTH {
482        return Err(MutationError::Malformed);
483    }
484    let key_length = usize::try_from(u32::from_le_bytes(copy_array(&encoded[1..5])))
485        .map_err(|_| MutationError::Malformed)?;
486    let key_end = DELETE_HEADER_LENGTH
487        .checked_add(key_length)
488        .ok_or(MutationError::Malformed)?;
489    if key_end != encoded.len() {
490        return Err(MutationError::Malformed);
491    }
492    let key = encoded[DELETE_HEADER_LENGTH..key_end].to_vec();
493    validate_key(&key)?;
494    validate_operation_length(encoded.len())?;
495    Ok(Mutation::Delete { key })
496}
497
498pub(crate) fn validate_key(key: &[u8]) -> Result<(), MutationError> {
499    if key.is_empty() {
500        return Err(MutationError::EmptyKey);
501    }
502    if key.len() > MAX_KEY_BYTES {
503        return Err(MutationError::KeyTooLarge {
504            length: key.len(),
505            maximum: MAX_KEY_BYTES,
506        });
507    }
508    Ok(())
509}
510
511fn validate_operation_length(length: usize) -> Result<(), MutationError> {
512    if length > MAX_OPERATION_BYTES {
513        return Err(MutationError::OperationTooLarge {
514            length,
515            maximum: MAX_OPERATION_BYTES,
516        });
517    }
518    Ok(())
519}
520
521fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
522    let mut output = [0_u8; N];
523    output.copy_from_slice(source);
524    output
525}
526
527#[cfg(test)]
528mod tests {
529    use std::error::Error;
530
531    use hyphae_core::{Q15Vector, VectorSpaceDefinition, VectorSpaceName};
532    use hyphae_query::FieldPath;
533    use hyphae_retrieval::{LexicalField, LexicalIndexDefinition};
534
535    use super::{MAX_KEY_BYTES, Mutation, MutationError};
536
537    #[test]
538    fn mutation_codec_round_trips_binary_values() -> Result<(), Box<dyn Error>> {
539        let mutations = [
540            Mutation::put([0, 1, 2], [255, 0, 3]),
541            Mutation::delete([7, 8, 9]),
542        ];
543        for mutation in mutations {
544            assert_eq!(Mutation::decode(&mutation.encode()?)?, mutation);
545        }
546        Ok(())
547    }
548
549    #[test]
550    fn vector_mutations_round_trip_canonically() -> Result<(), Box<dyn Error>> {
551        let space = VectorSpaceName::new("semantic.v1")?;
552        let mutations = [
553            Mutation::define_vector_space(VectorSpaceDefinition::cosine(space.clone(), 2)?),
554            Mutation::upsert_vector(space.clone(), b"object", Q15Vector::new(vec![32_767, -12])?),
555            Mutation::delete_vector(space, b"object"),
556        ];
557        for mutation in mutations {
558            assert_eq!(Mutation::decode(&mutation.encode()?)?, mutation);
559        }
560        Ok(())
561    }
562
563    #[test]
564    fn lexical_definition_mutation_round_trips_canonically() -> Result<(), Box<dyn Error>> {
565        let definition = LexicalIndexDefinition::new(
566            VectorSpaceName::new("documents")?,
567            vec![
568                LexicalField {
569                    path: FieldPath::new(["body", "text"]),
570                    weight_micros: 1_000_000,
571                },
572                LexicalField {
573                    path: FieldPath::field("title"),
574                    weight_micros: 2_000_000,
575                },
576            ],
577        )?;
578        let mutation = Mutation::define_lexical_index(definition);
579        assert_eq!(Mutation::decode(&mutation.encode()?)?, mutation);
580        Ok(())
581    }
582
583    #[test]
584    fn vector_mutation_decoder_rejects_invalid_q15_and_trailing_bytes() -> Result<(), Box<dyn Error>>
585    {
586        let space = VectorSpaceName::new("semantic")?;
587        let vector = Mutation::upsert_vector(space, b"object", Q15Vector::new(vec![1, 2])?);
588        let mut encoded = vector.encode()?;
589        encoded.push(0);
590        assert_eq!(Mutation::decode(&encoded), Err(MutationError::Malformed));
591        Ok(())
592    }
593
594    #[test]
595    fn mutation_codec_rejects_noncanonical_lengths() -> Result<(), Box<dyn Error>> {
596        let mut encoded = Mutation::put(b"key", b"value").encode()?;
597        encoded.push(0);
598        assert_eq!(Mutation::decode(&encoded), Err(MutationError::Malformed));
599        Ok(())
600    }
601
602    #[test]
603    fn keys_are_bounded_before_encoding() {
604        let result = Mutation::delete(vec![0; MAX_KEY_BYTES + 1]).encode();
605        assert!(matches!(result, Err(MutationError::KeyTooLarge { .. })));
606    }
607}
hyphae_storage/mutation.rs

hyphae_storage/
mutation.rs
