zeppelin_embed/ingest/
wal_payload.rs

1//! Versioned payloads inside the frozen task-08 WAL record frame.
2
3use crate::meta::{ColumnId, PredicateValue};
4
5use super::{DocId, DocumentVersion, IngestDocument, Revision};
6
7// Operation ids are an append-only persisted contract. Once assigned, an id
8// is never reused, even if its record kind is retired:
9//   1 = vector/document upsert v1
10//   2 = document delete v1
11//   3 = metadata edit v1
12//   4 = vector/document upsert with canonical ts v1
13//   5 = vector/document upsert with opaque stored metadata v1
14//   6 = vector/document upsert with canonical ts and opaque stored metadata v1
15//   7 = document upsert v2 with a u32 field-presence bitmap
16//
17// Upsert-v2 bitmap bits are append-only persisted meanings:
18//   bit 0 = vector (`dims:u32`, then `f32[dims]`)
19//   bit 1 = UTF-8 text (`len:u32`, then bytes)
20//   bit 2 = canonical timestamp (`i64`)
21//   bit 3 = opaque stored metadata (`len:u32`, then bytes)
22//   bit 4 = typed columns (count plus typed length-delimited entries)
23//   bit 5 = reserved-unused for a future per-record epoch tag
24//   bits 6..31 = reserved-unused
25/// Upsert payload v1.
26pub const UPSERT_V1: u16 = 1;
27/// Delete payload v1.
28pub const DELETE_V1: u16 = 2;
29/// Metadata-edit payload v1.
30pub const METADATA_EDIT_V1: u16 = 3;
31/// Upsert payload with canonical timestamp v1.
32pub const UPSERT_WITH_TIMESTAMP_V1: u16 = 4;
33/// Upsert payload with opaque stored metadata v1.
34pub const UPSERT_WITH_METADATA_V1: u16 = 5;
35/// Upsert payload with canonical timestamp and opaque stored metadata v1.
36pub const UPSERT_WITH_TIMESTAMP_AND_METADATA_V1: u16 = 6;
37/// Bitmap-based upsert payload v2.
38pub const UPSERT_V2: u16 = 7;
39
40/// Upsert-v2 vector field bit.
41pub const UPSERT_V2_VECTOR: u32 = 1 << 0;
42/// Upsert-v2 text field bit.
43pub const UPSERT_V2_TEXT: u32 = 1 << 1;
44/// Upsert-v2 timestamp field bit.
45pub const UPSERT_V2_TIMESTAMP: u32 = 1 << 2;
46/// Upsert-v2 opaque-metadata field bit.
47pub const UPSERT_V2_METADATA: u32 = 1 << 3;
48/// Upsert-v2 typed-column field bit.
49pub const UPSERT_V2_TYPED_COLUMNS: u32 = 1 << 4;
50/// Reserved-unused upsert-v2 bit for a future per-record epoch tag.
51pub const UPSERT_V2_EPOCH_TAG_RESERVED: u32 = 1 << 5;
52
53const UPSERT_V2_KNOWN_FIELDS: u32 = UPSERT_V2_VECTOR
54    | UPSERT_V2_TEXT
55    | UPSERT_V2_TIMESTAMP
56    | UPSERT_V2_METADATA
57    | UPSERT_V2_TYPED_COLUMNS;
58
59const PAYLOAD_VERSION: u16 = 1;
60const COMMON_HEADER_LEN: usize = 4;
61const DOCUMENT_VERSION_LEN: usize = 24;
62const UPSERT_PREFIX_LEN: usize = COMMON_HEADER_LEN + DOCUMENT_VERSION_LEN + 4;
63const TIMESTAMPED_UPSERT_PREFIX_LEN: usize = COMMON_HEADER_LEN + DOCUMENT_VERSION_LEN + 8 + 4;
64const METADATA_UPSERT_PREFIX_LEN: usize = COMMON_HEADER_LEN + DOCUMENT_VERSION_LEN + 4 + 4;
65const TIMESTAMPED_METADATA_UPSERT_PREFIX_LEN: usize =
66    COMMON_HEADER_LEN + DOCUMENT_VERSION_LEN + 8 + 4 + 4;
67const DELETE_PREFIX_LEN: usize = COMMON_HEADER_LEN + 4;
68const METADATA_PREFIX_LEN: usize = COMMON_HEADER_LEN + DOCUMENT_VERSION_LEN + 4 + 1 + 3 + 4;
69
70const VALUE_NULL: u8 = 0;
71const VALUE_U64: u8 = 1;
72const VALUE_I64: u8 = 2;
73const VALUE_F64: u8 = 3;
74const VALUE_BOOL: u8 = 4;
75const VALUE_STRING: u8 = 5;
76
77/// Typed operation-payload encoding or decoding failure.
78#[derive(Clone, Debug, Eq, PartialEq)]
79pub enum PayloadError {
80    /// A WAL operation id has no assigned payload contract.
81    UnknownOperation(u16),
82    /// A count or byte length does not fit its frozen field.
83    LengthOverflow,
84    /// The payload is shorter than the required fixed prefix or declared body.
85    Truncated,
86    /// The payload version is not v1.
87    Version(u16),
88    /// A must-be-zero flags field was non-zero.
89    Flags(u16),
90    /// A must-be-zero reserved metadata field was non-zero.
91    Reserved(u32),
92    /// An upsert contained no vector coordinates.
93    EmptyVector,
94    /// A delete contained no document ids.
95    EmptyDelete,
96    /// A vector coordinate was NaN or infinite.
97    NonFiniteVector {
98        /// Zero-based vector coordinate.
99        index: usize,
100    },
101    /// A metadata value kind has no assigned v1 meaning.
102    ValueKind(u8),
103    /// A value body has the wrong width for its declared kind.
104    ValueLength {
105        /// Declared metadata value kind.
106        kind: u8,
107        /// Required body width.
108        expected: usize,
109        /// Declared body width.
110        actual: usize,
111    },
112    /// A Boolean metadata byte was neither zero nor one.
113    Boolean(u8),
114    /// A string metadata body was not valid UTF-8.
115    Utf8,
116    /// Bytes remain after the declared payload.
117    TrailingBytes(usize),
118    /// An upsert-v2 bitmap named a reserved or unimplemented field.
119    FieldBitmap(u32),
120    /// Active lexical reconstruction failed during WAL replay.
121    Lexical(String),
122    /// Active typed-column reconstruction failed during WAL replay.
123    Columns(String),
124}
125
126impl std::fmt::Display for PayloadError {
127    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
128        match self {
129            Self::UnknownOperation(op) => {
130                write!(formatter, "WAL mutation operation {op} is unknown")
131            }
132            Self::LengthOverflow => formatter.write_str("WAL mutation payload length overflow"),
133            Self::Truncated => formatter.write_str("WAL mutation payload is truncated"),
134            Self::Version(version) => write!(
135                formatter,
136                "WAL mutation payload version {version} is unsupported"
137            ),
138            Self::Flags(flags) => write!(
139                formatter,
140                "WAL mutation payload flags {flags:#06x} are non-zero"
141            ),
142            Self::Reserved(value) => write!(
143                formatter,
144                "WAL metadata payload reserved bytes are {value:#08x}, expected zero"
145            ),
146            Self::EmptyVector => formatter.write_str("WAL upsert vector is empty"),
147            Self::EmptyDelete => formatter.write_str("WAL delete document list is empty"),
148            Self::NonFiniteVector { index } => write!(
149                formatter,
150                "WAL upsert vector coordinate {index} is non-finite"
151            ),
152            Self::ValueKind(kind) => {
153                write!(formatter, "WAL metadata value kind {kind} is unknown")
154            }
155            Self::ValueLength {
156                kind,
157                expected,
158                actual,
159            } => write!(
160                formatter,
161                "WAL metadata value kind {kind} needs {expected} bytes, got {actual}"
162            ),
163            Self::Boolean(value) => write!(
164                formatter,
165                "WAL metadata Boolean byte {value} is not canonical zero or one"
166            ),
167            Self::Utf8 => formatter.write_str("WAL metadata string is not valid UTF-8"),
168            Self::TrailingBytes(bytes) => {
169                write!(formatter, "WAL mutation payload has {bytes} trailing bytes")
170            }
171            Self::FieldBitmap(bitmap) => write!(
172                formatter,
173                "WAL upsert-v2 field bitmap {bitmap:#010x} names reserved fields"
174            ),
175            Self::Lexical(detail) => write!(formatter, "WAL lexical replay failed: {detail}"),
176            Self::Columns(detail) => write!(formatter, "WAL column replay failed: {detail}"),
177        }
178    }
179}
180
181impl std::error::Error for PayloadError {}
182
183/// Owned typed value carried by one metadata-edit record.
184#[derive(Clone, Debug, PartialEq)]
185pub enum MetadataValue {
186    /// Clears a nullable metadata value.
187    Null,
188    /// Unsigned integer value.
189    U64(u64),
190    /// Signed integer value.
191    I64(i64),
192    /// IEEE-754 value, preserving its exact bits.
193    F64(f64),
194    /// Boolean value.
195    Bool(bool),
196    /// UTF-8 string value.
197    String(String),
198}
199
200/// One structured document metadata mutation.
201#[derive(Clone, Debug, PartialEq)]
202pub struct MetadataEdit {
203    version: DocumentVersion,
204    column: ColumnId,
205    value: MetadataValue,
206}
207
208impl MetadataEdit {
209    /// Constructs one metadata edit for a specific document revision.
210    #[must_use]
211    pub const fn new(version: DocumentVersion, column: ColumnId, value: MetadataValue) -> Self {
212        Self {
213            version,
214            column,
215            value,
216        }
217    }
218
219    /// Returns the document/revision key to edit.
220    #[must_use]
221    pub const fn version(&self) -> DocumentVersion {
222        self.version
223    }
224
225    /// Returns the schema-local column id.
226    #[must_use]
227    pub const fn column(&self) -> ColumnId {
228        self.column
229    }
230
231    /// Returns the owned typed replacement value.
232    #[must_use]
233    pub const fn value(&self) -> &MetadataValue {
234        &self.value
235    }
236}
237
238/// One decoded record from the append-only mutation operation space.
239#[derive(Clone, Debug, PartialEq)]
240pub enum MutationPayload {
241    /// Vector/document upsert.
242    Upsert(IngestDocument),
243    /// Document tombstones.
244    Delete(Vec<DocId>),
245    /// Structured metadata edit.
246    MetadataEdit(MetadataEdit),
247}
248
249/// Encodes one upsert payload as little-endian
250/// `[version:u16, flags:u16, doc_id:u128, revision:u64, dims:u32, f32[dims]]`.
251pub fn encode_upsert(document: &IngestDocument) -> Result<Vec<u8>, PayloadError> {
252    encode_upsert_body(document, None, false)
253}
254
255/// Encodes one upsert with its canonical `ts` value before vector geometry.
256pub fn encode_upsert_with_timestamp(document: &IngestDocument) -> Result<Vec<u8>, PayloadError> {
257    encode_upsert_body(document, Some(document.timestamp()), false)
258}
259
260/// Encodes one upsert with opaque stored metadata.
261pub fn encode_upsert_with_metadata(document: &IngestDocument) -> Result<Vec<u8>, PayloadError> {
262    encode_upsert_body(document, None, true)
263}
264
265/// Encodes one timestamped upsert with opaque stored metadata.
266pub fn encode_upsert_with_timestamp_and_metadata(
267    document: &IngestDocument,
268) -> Result<Vec<u8>, PayloadError> {
269    encode_upsert_body(document, Some(document.timestamp()), true)
270}
271
272/// Encodes one bitmap-based upsert v2. The operation id carries the payload
273/// version, so the first four payload bytes are the field-presence bitmap.
274pub fn encode_upsert_v2(document: &IngestDocument) -> Result<Vec<u8>, PayloadError> {
275    let mut bitmap = UPSERT_V2_VECTOR;
276    if document.text().is_some() {
277        bitmap |= UPSERT_V2_TEXT;
278    }
279    if document.has_timestamp() {
280        bitmap |= UPSERT_V2_TIMESTAMP;
281    }
282    if document.has_metadata() {
283        bitmap |= UPSERT_V2_METADATA;
284    }
285    if document.has_columns() {
286        bitmap |= UPSERT_V2_TYPED_COLUMNS;
287    }
288    validate_vector(document.vector())?;
289    let dims = u32::try_from(document.vector().len()).map_err(|_| PayloadError::LengthOverflow)?;
290    let mut payload = Vec::new();
291    payload.extend_from_slice(&bitmap.to_le_bytes());
292    append_version(&mut payload, document.version());
293    payload.extend_from_slice(&dims.to_le_bytes());
294    for value in document.vector() {
295        payload.extend_from_slice(&value.to_bits().to_le_bytes());
296    }
297    if let Some(text) = document.text() {
298        append_length_prefixed(text.as_bytes(), &mut payload)?;
299    }
300    if document.has_timestamp() {
301        payload.extend_from_slice(&document.timestamp().to_le_bytes());
302    }
303    if document.has_metadata() {
304        append_length_prefixed(document.metadata(), &mut payload)?;
305    }
306    if document.has_columns() {
307        encode_typed_columns(document.columns(), &mut payload)?;
308    }
309    Ok(payload)
310}
311
312fn validate_vector(vector: &[f32]) -> Result<(), PayloadError> {
313    if vector.is_empty() {
314        return Err(PayloadError::EmptyVector);
315    }
316    if let Some(index) = vector.iter().position(|value| !value.is_finite()) {
317        return Err(PayloadError::NonFiniteVector { index });
318    }
319    Ok(())
320}
321
322fn append_length_prefixed(bytes: &[u8], output: &mut Vec<u8>) -> Result<(), PayloadError> {
323    let length = u32::try_from(bytes.len()).map_err(|_| PayloadError::LengthOverflow)?;
324    output.extend_from_slice(&length.to_le_bytes());
325    output.extend_from_slice(bytes);
326    Ok(())
327}
328
329fn encode_typed_columns(
330    columns: &[(ColumnId, PredicateValue)],
331    output: &mut Vec<u8>,
332) -> Result<(), PayloadError> {
333    let count = u32::try_from(columns.len()).map_err(|_| PayloadError::LengthOverflow)?;
334    output.extend_from_slice(&count.to_le_bytes());
335    for (column, value) in columns {
336        let metadata = predicate_to_metadata(value);
337        let (kind, body) = encode_metadata_value(&metadata)?;
338        output.extend_from_slice(&column.get().to_le_bytes());
339        output.push(kind);
340        output.extend_from_slice(&[0_u8; 3]);
341        append_length_prefixed(&body, output)?;
342    }
343    Ok(())
344}
345
346pub(crate) fn encode_column_values(
347    columns: &[(ColumnId, PredicateValue)],
348) -> Result<Vec<u8>, PayloadError> {
349    let mut encoded = Vec::new();
350    encode_typed_columns(columns, &mut encoded)?;
351    Ok(encoded)
352}
353
354fn predicate_to_metadata(value: &PredicateValue) -> MetadataValue {
355    match value {
356        PredicateValue::U64(value) => MetadataValue::U64(*value),
357        PredicateValue::I64(value) => MetadataValue::I64(*value),
358        PredicateValue::F64(value) => MetadataValue::F64(*value),
359        PredicateValue::Bool(value) => MetadataValue::Bool(*value),
360        PredicateValue::String(value) => MetadataValue::String(value.clone()),
361    }
362}
363
364fn encode_upsert_body(
365    document: &IngestDocument,
366    timestamp: Option<i64>,
367    include_metadata: bool,
368) -> Result<Vec<u8>, PayloadError> {
369    if document.vector().is_empty() {
370        return Err(PayloadError::EmptyVector);
371    }
372    let dims = u32::try_from(document.vector().len()).map_err(|_| PayloadError::LengthOverflow)?;
373    if let Some(index) = document
374        .vector()
375        .iter()
376        .position(|value| !value.is_finite())
377    {
378        return Err(PayloadError::NonFiniteVector { index });
379    }
380    let vector_bytes = document
381        .vector()
382        .len()
383        .checked_mul(std::mem::size_of::<f32>())
384        .ok_or(PayloadError::LengthOverflow)?;
385    let prefix = match (timestamp.is_some(), include_metadata) {
386        (false, false) => UPSERT_PREFIX_LEN,
387        (true, false) => TIMESTAMPED_UPSERT_PREFIX_LEN,
388        (false, true) => METADATA_UPSERT_PREFIX_LEN,
389        (true, true) => TIMESTAMPED_METADATA_UPSERT_PREFIX_LEN,
390    };
391    let capacity = prefix
392        .checked_add(if include_metadata {
393            document.metadata().len()
394        } else {
395            0
396        })
397        .and_then(|value| value.checked_add(vector_bytes))
398        .ok_or(PayloadError::LengthOverflow)?;
399    let mut payload = Vec::with_capacity(capacity);
400    append_header(&mut payload);
401    append_version(&mut payload, document.version());
402    if let Some(timestamp) = timestamp {
403        payload.extend_from_slice(&timestamp.to_le_bytes());
404    }
405    if include_metadata {
406        let metadata_len =
407            u32::try_from(document.metadata().len()).map_err(|_| PayloadError::LengthOverflow)?;
408        payload.extend_from_slice(&metadata_len.to_le_bytes());
409    }
410    payload.extend_from_slice(&dims.to_le_bytes());
411    if include_metadata {
412        payload.extend_from_slice(document.metadata());
413    }
414    for value in document.vector() {
415        payload.extend_from_slice(&value.to_bits().to_le_bytes());
416    }
417    Ok(payload)
418}
419
420/// Encodes one delete payload as little-endian
421/// `[version:u16, flags:u16, count:u32, doc_id:u128[count]]`.
422pub fn encode_delete(doc_ids: &[DocId]) -> Result<Vec<u8>, PayloadError> {
423    if doc_ids.is_empty() {
424        return Err(PayloadError::EmptyDelete);
425    }
426    let count = u32::try_from(doc_ids.len()).map_err(|_| PayloadError::LengthOverflow)?;
427    let body_bytes = doc_ids
428        .len()
429        .checked_mul(std::mem::size_of::<u128>())
430        .ok_or(PayloadError::LengthOverflow)?;
431    let capacity = DELETE_PREFIX_LEN
432        .checked_add(body_bytes)
433        .ok_or(PayloadError::LengthOverflow)?;
434    let mut payload = Vec::with_capacity(capacity);
435    append_header(&mut payload);
436    payload.extend_from_slice(&count.to_le_bytes());
437    for doc_id in doc_ids {
438        payload.extend_from_slice(&doc_id.get().to_le_bytes());
439    }
440    Ok(payload)
441}
442
443/// Encodes one metadata edit as little-endian fixed identity/type fields plus
444/// a length-delimited value body.
445pub fn encode_metadata_edit(edit: &MetadataEdit) -> Result<Vec<u8>, PayloadError> {
446    let (kind, body) = encode_metadata_value(edit.value())?;
447    let value_len = u32::try_from(body.len()).map_err(|_| PayloadError::LengthOverflow)?;
448    let capacity = METADATA_PREFIX_LEN
449        .checked_add(body.len())
450        .ok_or(PayloadError::LengthOverflow)?;
451    let mut payload = Vec::with_capacity(capacity);
452    append_header(&mut payload);
453    append_version(&mut payload, edit.version());
454    payload.extend_from_slice(&edit.column().get().to_le_bytes());
455    payload.push(kind);
456    payload.extend_from_slice(&[0_u8; 3]);
457    payload.extend_from_slice(&value_len.to_le_bytes());
458    payload.extend_from_slice(&body);
459    Ok(payload)
460}
461
462/// Decodes a payload according to its append-only operation id.
463pub fn decode_mutation(op: u16, payload: &[u8]) -> Result<MutationPayload, PayloadError> {
464    match op {
465        UPSERT_V1 => decode_upsert(payload).map(MutationPayload::Upsert),
466        DELETE_V1 => decode_delete(payload).map(MutationPayload::Delete),
467        METADATA_EDIT_V1 => decode_metadata_edit(payload).map(MutationPayload::MetadataEdit),
468        UPSERT_WITH_TIMESTAMP_V1 => {
469            decode_upsert_with_timestamp(payload).map(MutationPayload::Upsert)
470        }
471        UPSERT_WITH_METADATA_V1 => {
472            decode_upsert_with_metadata(payload).map(MutationPayload::Upsert)
473        }
474        UPSERT_WITH_TIMESTAMP_AND_METADATA_V1 => {
475            decode_upsert_with_timestamp_and_metadata(payload).map(MutationPayload::Upsert)
476        }
477        UPSERT_V2 => decode_upsert_v2(payload).map(MutationPayload::Upsert),
478        unknown => Err(PayloadError::UnknownOperation(unknown)),
479    }
480}
481
482/// Decodes one bitmap-based upsert-v2 payload.
483pub fn decode_upsert_v2(payload: &[u8]) -> Result<IngestDocument, PayloadError> {
484    let mut cursor = Cursor::new(payload);
485    let bitmap = cursor.read_u32()?;
486    let reserved = bitmap & !UPSERT_V2_KNOWN_FIELDS;
487    if reserved != 0 {
488        return Err(PayloadError::FieldBitmap(reserved));
489    }
490    let version = cursor.read_version()?;
491    let vector = if bitmap & UPSERT_V2_VECTOR != 0 {
492        let dims = usize::try_from(cursor.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?;
493        if dims == 0 {
494            return Err(PayloadError::EmptyVector);
495        }
496        let mut vector = Vec::with_capacity(dims);
497        for index in 0..dims {
498            let value = f32::from_bits(cursor.read_u32()?);
499            if !value.is_finite() {
500                return Err(PayloadError::NonFiniteVector { index });
501            }
502            vector.push(value);
503        }
504        vector
505    } else {
506        Vec::new()
507    };
508    let text = if bitmap & UPSERT_V2_TEXT != 0 {
509        Some(cursor.read_string()?)
510    } else {
511        None
512    };
513    let timestamp = if bitmap & UPSERT_V2_TIMESTAMP != 0 {
514        Some(i64::from_le_bytes(cursor.take()?))
515    } else {
516        None
517    };
518    let metadata = if bitmap & UPSERT_V2_METADATA != 0 {
519        Some(cursor.read_length_prefixed()?.to_vec())
520    } else {
521        None
522    };
523    let columns = if bitmap & UPSERT_V2_TYPED_COLUMNS != 0 {
524        Some(decode_typed_columns(&mut cursor)?)
525    } else {
526        None
527    };
528    cursor.finish()?;
529    let mut document = IngestDocument::new(version, vector);
530    if let Some(text) = text {
531        document = document.with_text(text);
532    }
533    if let Some(timestamp) = timestamp {
534        document = document.with_timestamp(timestamp);
535    }
536    if let Some(metadata) = metadata {
537        document = document.with_metadata(metadata);
538    }
539    if let Some(columns) = columns {
540        document = document.with_columns(columns);
541    }
542    Ok(document)
543}
544
545fn decode_typed_columns(
546    cursor: &mut Cursor<'_>,
547) -> Result<Vec<(ColumnId, PredicateValue)>, PayloadError> {
548    let count = usize::try_from(cursor.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?;
549    let mut columns = Vec::with_capacity(count);
550    for _ in 0..count {
551        let column = ColumnId::new(cursor.read_u32()?);
552        let kind = cursor.read_u8()?;
553        let reserved = cursor.read_u24()?;
554        if reserved != 0 {
555            return Err(PayloadError::Reserved(reserved));
556        }
557        let body = cursor.read_length_prefixed()?;
558        let value = metadata_to_predicate(decode_metadata_value(kind, body)?)?;
559        columns.push((column, value));
560    }
561    Ok(columns)
562}
563
564pub(crate) fn decode_column_values(
565    encoded: &[u8],
566) -> Result<Vec<(ColumnId, PredicateValue)>, PayloadError> {
567    let mut cursor = Cursor::new(encoded);
568    let columns = decode_typed_columns(&mut cursor)?;
569    cursor.finish()?;
570    Ok(columns)
571}
572
573fn metadata_to_predicate(value: MetadataValue) -> Result<PredicateValue, PayloadError> {
574    match value {
575        MetadataValue::Null => Err(PayloadError::ValueKind(VALUE_NULL)),
576        MetadataValue::U64(value) => Ok(PredicateValue::U64(value)),
577        MetadataValue::I64(value) => Ok(PredicateValue::I64(value)),
578        MetadataValue::F64(value) => Ok(PredicateValue::F64(value)),
579        MetadataValue::Bool(value) => Ok(PredicateValue::Bool(value)),
580        MetadataValue::String(value) => Ok(PredicateValue::String(value)),
581    }
582}
583
584/// Decodes and validates one complete upsert payload.
585pub fn decode_upsert(payload: &[u8]) -> Result<IngestDocument, PayloadError> {
586    decode_upsert_body(payload, false, false)
587}
588
589/// Decodes one complete upsert carrying the canonical `ts` value.
590pub fn decode_upsert_with_timestamp(payload: &[u8]) -> Result<IngestDocument, PayloadError> {
591    decode_upsert_body(payload, true, false)
592}
593
594/// Decodes one complete upsert carrying opaque stored metadata.
595pub fn decode_upsert_with_metadata(payload: &[u8]) -> Result<IngestDocument, PayloadError> {
596    decode_upsert_body(payload, false, true)
597}
598
599/// Decodes one complete timestamped upsert carrying opaque stored metadata.
600pub fn decode_upsert_with_timestamp_and_metadata(
601    payload: &[u8],
602) -> Result<IngestDocument, PayloadError> {
603    decode_upsert_body(payload, true, true)
604}
605
606fn decode_upsert_body(
607    payload: &[u8],
608    has_timestamp: bool,
609    has_metadata: bool,
610) -> Result<IngestDocument, PayloadError> {
611    let mut cursor = Cursor::new(payload);
612    cursor.require_header()?;
613    let version = cursor.read_version()?;
614    let timestamp = if has_timestamp {
615        i64::from_le_bytes(cursor.take()?)
616    } else {
617        0
618    };
619    let metadata_len = if has_metadata {
620        usize::try_from(cursor.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?
621    } else {
622        0
623    };
624    let dims = usize::try_from(cursor.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?;
625    if dims == 0 {
626        return Err(PayloadError::EmptyVector);
627    }
628    let body_bytes = dims
629        .checked_mul(std::mem::size_of::<f32>())
630        .ok_or(PayloadError::LengthOverflow)?;
631    let remaining = metadata_len
632        .checked_add(body_bytes)
633        .ok_or(PayloadError::LengthOverflow)?;
634    cursor.require_remaining(remaining)?;
635    let metadata = cursor.read_bytes(metadata_len)?.to_vec();
636    let mut vector = Vec::with_capacity(dims);
637    for index in 0..dims {
638        let value = f32::from_bits(cursor.read_u32()?);
639        if !value.is_finite() {
640            return Err(PayloadError::NonFiniteVector { index });
641        }
642        vector.push(value);
643    }
644    cursor.finish()?;
645    let mut document = IngestDocument::new(version, vector);
646    if has_timestamp {
647        document = document.with_timestamp(timestamp);
648    }
649    if has_metadata {
650        document = document.with_metadata(metadata);
651    }
652    Ok(document)
653}
654
655/// Decodes and validates one complete delete payload.
656pub fn decode_delete(payload: &[u8]) -> Result<Vec<DocId>, PayloadError> {
657    let mut cursor = Cursor::new(payload);
658    cursor.require_header()?;
659    let count = usize::try_from(cursor.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?;
660    if count == 0 {
661        return Err(PayloadError::EmptyDelete);
662    }
663    let body_bytes = count
664        .checked_mul(std::mem::size_of::<u128>())
665        .ok_or(PayloadError::LengthOverflow)?;
666    cursor.require_remaining(body_bytes)?;
667    let mut doc_ids = Vec::with_capacity(count);
668    for _ in 0..count {
669        doc_ids.push(DocId::new(cursor.read_u128()?));
670    }
671    cursor.finish()?;
672    Ok(doc_ids)
673}
674
675/// Decodes and validates one complete metadata-edit payload.
676pub fn decode_metadata_edit(payload: &[u8]) -> Result<MetadataEdit, PayloadError> {
677    let mut cursor = Cursor::new(payload);
678    cursor.require_header()?;
679    let version = cursor.read_version()?;
680    let column = ColumnId::new(cursor.read_u32()?);
681    let kind = cursor.read_u8()?;
682    let reserved = cursor.read_u24()?;
683    if reserved != 0 {
684        return Err(PayloadError::Reserved(reserved));
685    }
686    let value_len =
687        usize::try_from(cursor.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?;
688    let body = cursor.read_bytes(value_len)?;
689    let value = decode_metadata_value(kind, body)?;
690    cursor.finish()?;
691    Ok(MetadataEdit::new(version, column, value))
692}
693
694fn append_header(payload: &mut Vec<u8>) {
695    payload.extend_from_slice(&PAYLOAD_VERSION.to_le_bytes());
696    payload.extend_from_slice(&0_u16.to_le_bytes());
697}
698
699fn append_version(payload: &mut Vec<u8>, version: DocumentVersion) {
700    payload.extend_from_slice(&version.doc_id().get().to_le_bytes());
701    payload.extend_from_slice(&version.revision().get().to_le_bytes());
702}
703
704fn encode_metadata_value(value: &MetadataValue) -> Result<(u8, Vec<u8>), PayloadError> {
705    let encoded = match value {
706        MetadataValue::Null => (VALUE_NULL, Vec::new()),
707        MetadataValue::U64(value) => (VALUE_U64, value.to_le_bytes().to_vec()),
708        MetadataValue::I64(value) => (VALUE_I64, value.to_le_bytes().to_vec()),
709        MetadataValue::F64(value) => (VALUE_F64, value.to_bits().to_le_bytes().to_vec()),
710        MetadataValue::Bool(value) => (VALUE_BOOL, vec![u8::from(*value)]),
711        MetadataValue::String(value) => {
712            u32::try_from(value.len()).map_err(|_| PayloadError::LengthOverflow)?;
713            (VALUE_STRING, value.as_bytes().to_vec())
714        }
715    };
716    Ok(encoded)
717}
718
719fn decode_metadata_value(kind: u8, body: &[u8]) -> Result<MetadataValue, PayloadError> {
720    match kind {
721        VALUE_NULL => {
722            require_value_len(kind, body, 0)?;
723            Ok(MetadataValue::Null)
724        }
725        VALUE_U64 => {
726            require_value_len(kind, body, 8)?;
727            read_array(body)
728                .map(u64::from_le_bytes)
729                .map(MetadataValue::U64)
730        }
731        VALUE_I64 => {
732            require_value_len(kind, body, 8)?;
733            read_array(body)
734                .map(i64::from_le_bytes)
735                .map(MetadataValue::I64)
736        }
737        VALUE_F64 => {
738            require_value_len(kind, body, 8)?;
739            read_array(body)
740                .map(u64::from_le_bytes)
741                .map(f64::from_bits)
742                .map(MetadataValue::F64)
743        }
744        VALUE_BOOL => {
745            require_value_len(kind, body, 1)?;
746            match body.first().copied().ok_or(PayloadError::Truncated)? {
747                0 => Ok(MetadataValue::Bool(false)),
748                1 => Ok(MetadataValue::Bool(true)),
749                value => Err(PayloadError::Boolean(value)),
750            }
751        }
752        VALUE_STRING => std::str::from_utf8(body)
753            .map(|value| MetadataValue::String(value.to_owned()))
754            .map_err(|_| PayloadError::Utf8),
755        unknown => Err(PayloadError::ValueKind(unknown)),
756    }
757}
758
759fn require_value_len(kind: u8, body: &[u8], expected: usize) -> Result<(), PayloadError> {
760    if body.len() == expected {
761        Ok(())
762    } else {
763        Err(PayloadError::ValueLength {
764            kind,
765            expected,
766            actual: body.len(),
767        })
768    }
769}
770
771fn read_array<const N: usize>(bytes: &[u8]) -> Result<[u8; N], PayloadError> {
772    <[u8; N]>::try_from(bytes).map_err(|_| PayloadError::Truncated)
773}
774
775struct Cursor<'a> {
776    bytes: &'a [u8],
777    offset: usize,
778}
779
780impl<'a> Cursor<'a> {
781    const fn new(bytes: &'a [u8]) -> Self {
782        Self { bytes, offset: 0 }
783    }
784
785    fn require_header(&mut self) -> Result<(), PayloadError> {
786        let version = self.read_u16()?;
787        if version != PAYLOAD_VERSION {
788            return Err(PayloadError::Version(version));
789        }
790        let flags = self.read_u16()?;
791        if flags != 0 {
792            return Err(PayloadError::Flags(flags));
793        }
794        Ok(())
795    }
796
797    fn read_version(&mut self) -> Result<DocumentVersion, PayloadError> {
798        Ok(DocumentVersion::new(
799            DocId::new(self.read_u128()?),
800            Revision::new(self.read_u64()?),
801        ))
802    }
803
804    fn take<const N: usize>(&mut self) -> Result<[u8; N], PayloadError> {
805        read_array(self.read_bytes(N)?)
806    }
807
808    fn read_bytes(&mut self, length: usize) -> Result<&'a [u8], PayloadError> {
809        let end = self
810            .offset
811            .checked_add(length)
812            .ok_or(PayloadError::LengthOverflow)?;
813        let bytes = self
814            .bytes
815            .get(self.offset..end)
816            .ok_or(PayloadError::Truncated)?;
817        self.offset = end;
818        Ok(bytes)
819    }
820
821    fn require_remaining(&self, length: usize) -> Result<(), PayloadError> {
822        if self.remaining() < length {
823            Err(PayloadError::Truncated)
824        } else {
825            Ok(())
826        }
827    }
828
829    fn read_u8(&mut self) -> Result<u8, PayloadError> {
830        self.take::<1>().map(u8::from_le_bytes)
831    }
832
833    fn read_u16(&mut self) -> Result<u16, PayloadError> {
834        self.take().map(u16::from_le_bytes)
835    }
836
837    fn read_u24(&mut self) -> Result<u32, PayloadError> {
838        let [low, middle, high] = self.take::<3>()?;
839        Ok(u32::from(low) | (u32::from(middle) << 8) | (u32::from(high) << 16))
840    }
841
842    fn read_u32(&mut self) -> Result<u32, PayloadError> {
843        self.take().map(u32::from_le_bytes)
844    }
845
846    fn read_u64(&mut self) -> Result<u64, PayloadError> {
847        self.take().map(u64::from_le_bytes)
848    }
849
850    fn read_u128(&mut self) -> Result<u128, PayloadError> {
851        self.take().map(u128::from_le_bytes)
852    }
853
854    fn read_length_prefixed(&mut self) -> Result<&'a [u8], PayloadError> {
855        let length = usize::try_from(self.read_u32()?).map_err(|_| PayloadError::LengthOverflow)?;
856        self.read_bytes(length)
857    }
858
859    fn read_string(&mut self) -> Result<String, PayloadError> {
860        std::str::from_utf8(self.read_length_prefixed()?)
861            .map(str::to_owned)
862            .map_err(|_| PayloadError::Utf8)
863    }
864
865    fn remaining(&self) -> usize {
866        self.bytes.len().saturating_sub(self.offset)
867    }
868
869    fn finish(self) -> Result<(), PayloadError> {
870        let remaining = self.remaining();
871        if remaining == 0 {
872            Ok(())
873        } else {
874            Err(PayloadError::TrailingBytes(remaining))
875        }
876    }
877}
zeppelin_embed/ingest/wal_payload.rs

zeppelin_embed/ingest/
wal_payload.rs
