lance_core/cache/
codec.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4//! Serialization codecs for cache entries.
5//!
6//! Implement [`CacheCodecImpl`] on concrete types, then use
7//! [`CacheCodec::from_impl`] to produce a type-erased codec for the cache.
8//!
9//! # Wire format
10//!
11//! Every serialized entry begins with a small hand-framed **envelope** so the
12//! reader can validate it before trusting the body:
13//!
14//! ```text
15//! [magic: 4B = b"LCE1"]
16//! [envelope_version: u8]
17//! [type_id_len: u16 LE][type_id: utf8]   # stable, author-assigned
18//! [type_version: u32 LE]                 # per-type body schema version
19//! <body, written by the type's CacheCodecImpl::serialize>
20//! ```
21//!
22//! The envelope is deliberately *not* protobuf: it is the most
23//! stability-critical part, must parse robustly against arbitrary bytes
24//! (including data written by older, pre-stabilization builds), and never
25//! changes shape. Bodies use protobuf headers, where field-number evolution
26//! pays off.
27//!
28//! # Decode outcome
29//!
30//! Deserialization never propagates a parse failure as a hard error into the
31//! cache path. Anything the reader cannot confidently interpret — absent or
32//! wrong magic, an unknown `envelope_version`, a `type_id` mismatch, an
33//! unsupported `type_version`, or a body decode error — becomes
34//! [`CacheDecode::Miss`]. A backend turns `Miss` into a normal cache miss and
35//! recomputes the value. This is what lets data written by an older format
36//! self-heal: it simply fails the magic check and is regenerated.
37
38use std::io::Write;
39use std::sync::Arc;
40
41use bytes::Bytes;
42
43use crate::{Error, Result};
44
45use super::{CacheEntryReader, CacheEntryWriter};
46
47// ---------------------------------------------------------------------------
48// Envelope
49// ---------------------------------------------------------------------------
50
51/// Magic bytes that prefix every stabilized cache entry.
52///
53/// An ASCII tag (`0x4C 0x43 0x45 0x31`) chosen so it cannot collide with any
54/// pre-stabilization blob: those began with either a small little-endian
55/// length (tens of bytes) or a small tag byte, never these values.
56///
57/// Exported so backends can cheaply identify Lance cache entries (e.g. when
58/// scanning a persistent store at startup) without hardcoding the bytes —
59/// prefer [`has_cache_envelope`] over comparing against this directly.
60pub const MAGIC: [u8; 4] = *b"LCE1";
61
62/// Returns `true` if `data` begins with the cache-entry [`MAGIC`].
63///
64/// A cheap prefix check for backends that need to recognize Lance cache
65/// entries without fully [`deserialize`](CacheCodec::deserialize)-ing them. A
66/// `true` result only means the framing looks like ours; the entry can still
67/// decode to a [`Miss`](CacheDecode::Miss) (e.g. wrong `type_id`).
68pub fn has_cache_envelope(data: &[u8]) -> bool {
69    data.get(..MAGIC.len()) == Some(&MAGIC[..])
70}
71
72/// Version of the envelope framing itself. Bumped only if the outer frame
73/// (magic/version/type_id/type_version layout) ever changes — expected never.
74const ENVELOPE_VERSION: u8 = 1;
75
76/// Parsed envelope borrowed from the input bytes.
77struct ParsedEnvelope<'a> {
78    type_id: &'a str,
79    type_version: u32,
80    /// Offset of the first body byte within the input.
81    body_offset: usize,
82}
83
84/// Parse and validate the envelope at the start of `data`.
85///
86/// Returns `None` for anything that is not a well-formed envelope this build
87/// understands (wrong/absent magic, unknown `envelope_version`, truncation,
88/// non-utf8 `type_id`). Callers translate `None` into [`CacheDecode::Miss`].
89fn parse_envelope(data: &Bytes) -> Option<ParsedEnvelope<'_>> {
90    let bytes = data.as_ref();
91    let mut off = 0usize;
92
93    let magic = bytes.get(off..off + 4)?;
94    if magic != MAGIC {
95        return None;
96    }
97    off += 4;
98
99    if *bytes.get(off)? != ENVELOPE_VERSION {
100        return None;
101    }
102    off += 1;
103
104    let type_id_len = u16::from_le_bytes(bytes.get(off..off + 2)?.try_into().ok()?) as usize;
105    off += 2;
106
107    let type_id = std::str::from_utf8(bytes.get(off..off + type_id_len)?).ok()?;
108    off += type_id_len;
109
110    let type_version = u32::from_le_bytes(bytes.get(off..off + 4)?.try_into().ok()?);
111    off += 4;
112
113    Some(ParsedEnvelope {
114        type_id,
115        type_version,
116        body_offset: off,
117    })
118}
119
120/// Write the envelope for `type_id`/`type_version`, returning the number of
121/// bytes written (the body's starting offset).
122fn write_envelope(writer: &mut dyn Write, type_id: &str, type_version: u32) -> Result<usize> {
123    let type_id_len = u16::try_from(type_id.len()).map_err(|_| {
124        Error::io(format!(
125            "cache codec type_id too long ({} bytes, max {})",
126            type_id.len(),
127            u16::MAX
128        ))
129    })?;
130
131    writer.write_all(&MAGIC)?;
132    writer.write_all(&[ENVELOPE_VERSION])?;
133    writer.write_all(&type_id_len.to_le_bytes())?;
134    writer.write_all(type_id.as_bytes())?;
135    writer.write_all(&type_version.to_le_bytes())?;
136
137    Ok(4 + 1 + 2 + type_id.len() + 4)
138}
139
140// ---------------------------------------------------------------------------
141// CacheDecode — first-class cache-miss outcome
142// ---------------------------------------------------------------------------
143
144/// Why a cache entry could not be decoded into the expected type.
145///
146/// Carried by [`CacheDecode::Miss`] so backends can emit targeted metrics
147/// (e.g. distinguish "evicting due to a stale format" from "type collision")
148/// without re-parsing. Every reason maps to the same behavior — recompute via
149/// the loader — so callers that don't care can ignore it.
150#[derive(Debug, Clone, Copy, PartialEq, Eq)]
151pub enum CacheMissReason {
152    /// Absent or wrong magic, unknown `envelope_version`, truncated framing, or
153    /// a non-utf8 `type_id`. Typically an entry written by a pre-stabilization
154    /// or otherwise foreign build.
155    InvalidEnvelope,
156    /// Well-formed envelope, but its `type_id` names a different entry type than
157    /// the codec reading it.
158    TypeMismatch,
159    /// Written by a newer build whose `type_version` this build does not
160    /// understand and must not attempt to interpret.
161    VersionTooNew,
162    /// Envelope validated, but the body failed to decode (truncation, a
163    /// malformed protobuf header, an IPC error, etc.).
164    BodyError,
165}
166
167/// Outcome of deserializing a cache entry.
168///
169/// `Miss` means the bytes could not be confidently decoded into `T`; the
170/// [`CacheMissReason`] says why. A backend treats any `Miss` exactly like a key
171/// that was never present: recompute via the loader.
172#[derive(Debug)]
173pub enum CacheDecode<T> {
174    Hit(T),
175    Miss(CacheMissReason),
176}
177
178impl<T> CacheDecode<T> {
179    pub fn hit(self) -> Option<T> {
180        match self {
181            Self::Hit(v) => Some(v),
182            Self::Miss(_) => None,
183        }
184    }
185}
186
187// ---------------------------------------------------------------------------
188// CacheCodecImpl — trait for serializable cache entry types
189// ---------------------------------------------------------------------------
190
191/// Bounded random reads from a persistent cache payload. Backends may coalesce
192/// and align physical I/O while codecs request logical byte ranges.
193pub trait CacheRangeReader {
194    fn read_range(&self, range: std::ops::Range<usize>) -> Result<Bytes>;
195}
196impl<F: Fn(std::ops::Range<usize>) -> Result<Bytes>> CacheRangeReader for F {
197    fn read_range(&self, range: std::ops::Range<usize>) -> Result<Bytes> {
198        self(range)
199    }
200}
201
202/// Serialization trait for cache entries.
203///
204/// **Experimental**: the serialized format is not yet covered by a stability
205/// guarantee and may change between releases. When it does stabilize, the
206/// rules are: `TYPE_ID`, protobuf field numbers, and enum values are
207/// append-only forever; format changes that protobuf cannot express
208/// transparently bump [`CURRENT_VERSION`](Self::CURRENT_VERSION).
209///
210/// Implement this on concrete types that need to survive serialization through
211/// a persistent cache backend, then wire it into a
212/// [`CacheKey`](super::CacheKey) via [`CacheCodec::from_impl`].
213///
214/// The envelope (magic/version/type_id/type_version) is written and validated
215/// by the [`CacheCodec`] wrapper. [`serialize`](Self::serialize) writes only
216/// the body — a header followed by sections in a fixed, version-keyed order —
217/// and [`deserialize`](Self::deserialize) reads them back in that same order.
218/// The read sequence mirroring the write sequence for each `type_version` is
219/// the invariant the implementor owns.
220pub trait CacheCodecImpl: Send + Sync {
221    /// Whether this body supports gathering sorted, unique row offsets.
222    const SUPPORTS_ROW_SELECTION: bool = false;
223    fn deserialize_rows(
224        _reader: &dyn CacheRangeReader,
225        _body_offset: usize,
226        _version: u32,
227        _rows: &[u32],
228    ) -> Result<Self>
229    where
230        Self: Sized,
231    {
232        Err(Error::invalid_input(
233            "cache codec does not support row selection",
234        ))
235    }
236
237    /// Payload byte ranges that [`deserialize_rows`](Self::deserialize_rows)
238    /// requests for `rows`, in request order, excluding the envelope and any
239    /// body header it reads through `reader` while planning.
240    ///
241    /// Lets a backend issue those reads concurrently before decoding. The
242    /// default `None` means the body cannot plan, and backends read serially.
243    fn plan_row_ranges(
244        _reader: &dyn CacheRangeReader,
245        _body_offset: usize,
246        _version: u32,
247        _rows: &[u32],
248    ) -> Result<Option<Vec<std::ops::Range<usize>>>>
249    where
250        Self: Sized,
251    {
252        Ok(None)
253    }
254
255    /// Stable identity for this entry type. **Must not change once shipped.**
256    /// This is a deliberate author-assigned string, not `std::any::type_name`
257    /// (which is not stable across compiler versions).
258    const TYPE_ID: &'static str;
259
260    /// Body schema version this build writes. Bump when the body layout
261    /// changes in a way protobuf field additions cannot express transparently
262    /// (adding/removing/reordering sections, a raw-blob encoding change, etc.).
263    const CURRENT_VERSION: u32;
264
265    /// Write the body: a header, then sections in a fixed order.
266    fn serialize(&self, writer: &mut CacheEntryWriter<'_>) -> Result<()>;
267
268    /// Reconstruct from the body. Branch on
269    /// [`reader.version()`](CacheEntryReader::version) for backward compat;
270    /// sections are read in write order.
271    fn deserialize(reader: &mut CacheEntryReader<'_>) -> Result<Self>
272    where
273        Self: Sized;
274}
275
276// ---------------------------------------------------------------------------
277// CacheCodec — type-erased codec passed to backends
278// ---------------------------------------------------------------------------
279
280pub(crate) type ArcAny = Arc<dyn std::any::Any + Send + Sync>;
281
282type RowDecoder = fn(&dyn CacheRangeReader, usize, u32, &[u32]) -> Result<ArcAny>;
283type RowPlanner =
284    fn(&dyn CacheRangeReader, usize, u32, &[u32]) -> Result<Option<Vec<std::ops::Range<usize>>>>;
285
286/// Envelope bytes before `type_id`: magic, envelope version and `type_id_len`.
287const ENVELOPE_ID_OFFSET: usize = 4 + 1 + 2;
288/// Envelope bytes that follow `type_id`: `type_version`.
289const ENVELOPE_VERSION_BYTES: usize = 4;
290
291/// Type-erased codec for serializing and deserializing cache entries.
292///
293/// `CacheCodec` carries the entry's stable `type_id`/`version` plus plain
294/// function pointers — it is `Copy` and has no heap allocation. Construct one
295/// via [`CacheCodec::from_impl`] for types that implement [`CacheCodecImpl`],
296/// or [`CacheCodec::new`] for custom cases (e.g. when the orphan rule prevents
297/// a direct impl).
298#[derive(Copy, Clone)]
299pub struct CacheCodec {
300    row_decoder: Option<RowDecoder>,
301    row_planner: Option<RowPlanner>,
302    memory_priority: u8,
303    plane_tag: Option<u8>,
304    type_id: &'static str,
305    version: u32,
306    serialize_body: fn(&ArcAny, &mut CacheEntryWriter<'_>) -> Result<()>,
307    deserialize_body: fn(&mut CacheEntryReader<'_>) -> Result<ArcAny>,
308}
309
310impl std::fmt::Debug for CacheCodec {
311    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
312        f.debug_struct("CacheCodec")
313            .field("type_id", &self.type_id)
314            .field("version", &self.version)
315            .finish_non_exhaustive()
316    }
317}
318
319fn serialize_via_impl<T: CacheCodecImpl + 'static>(
320    any: &ArcAny,
321    writer: &mut CacheEntryWriter<'_>,
322) -> Result<()> {
323    let val = any
324        .downcast_ref::<T>()
325        .expect("CacheCodec::serialize called with wrong type (this is a bug in the cache layer)");
326    val.serialize(writer)
327}
328
329fn deserialize_via_impl<T: CacheCodecImpl + 'static>(
330    reader: &mut CacheEntryReader<'_>,
331) -> Result<ArcAny> {
332    let val = T::deserialize(reader)?;
333    Ok(Arc::new(val) as ArcAny)
334}
335
336impl CacheCodec {
337    /// Create a `CacheCodec` from explicit body function pointers.
338    ///
339    /// Prefer [`from_impl`](Self::from_impl) when the value type implements
340    /// [`CacheCodecImpl`]. Use this for types where a direct impl isn't
341    /// possible (e.g. the orphan rule prevents it). `type_id` and `version`
342    /// play the same role as the corresponding [`CacheCodecImpl`] constants.
343    pub fn new(
344        type_id: &'static str,
345        version: u32,
346        serialize_body: fn(&ArcAny, &mut CacheEntryWriter<'_>) -> Result<()>,
347        deserialize_body: fn(&mut CacheEntryReader<'_>) -> Result<ArcAny>,
348    ) -> Self {
349        Self {
350            row_decoder: None,
351            row_planner: None,
352            memory_priority: 0,
353            plane_tag: None,
354            type_id,
355            version,
356            serialize_body,
357            deserialize_body,
358        }
359    }
360
361    /// Create a `CacheCodec` from a [`CacheCodecImpl`] implementation.
362    pub fn from_impl<T: CacheCodecImpl + 'static>() -> Self {
363        Self {
364            row_decoder: if T::SUPPORTS_ROW_SELECTION {
365                Some(|reader, offset, version, rows| {
366                    Ok(Arc::new(T::deserialize_rows(reader, offset, version, rows)?) as ArcAny)
367                })
368            } else {
369                None
370            },
371            row_planner: if T::SUPPORTS_ROW_SELECTION {
372                Some(T::plan_row_ranges)
373            } else {
374                None
375            },
376            memory_priority: 0,
377            plane_tag: None,
378            type_id: T::TYPE_ID,
379            version: T::CURRENT_VERSION,
380            serialize_body: serialize_via_impl::<T>,
381            deserialize_body: deserialize_via_impl::<T>,
382        }
383    }
384
385    /// Hint for priority-aware memory tiers. Larger values survive pressure first.
386    pub fn with_memory_priority(mut self, priority: u8) -> Self {
387        self.memory_priority = priority;
388        self
389    }
390
391    pub fn memory_priority(&self) -> u8 {
392        self.memory_priority
393    }
394
395    /// Tag the entry as one plane of a layered index so backends can report
396    /// per-plane metrics without deriving the plane from an admission priority.
397    pub fn with_plane_tag(mut self, plane: u8) -> Self {
398        self.plane_tag = Some(plane);
399        self
400    }
401
402    /// The layered plane this entry belongs to, or `None` for other entries.
403    pub fn plane_tag(&self) -> Option<u8> {
404        self.plane_tag
405    }
406
407    /// Return the stable entry type identity.
408    ///
409    /// Persistent and tiered backends can use this metadata to route encoded
410    /// values without retaining a readable logical cache key.
411    pub const fn type_id(&self) -> &'static str {
412        self.type_id
413    }
414
415    /// Serialize `value` into `writer`: envelope first, then the body.
416    pub fn serialize(&self, value: &ArcAny, writer: &mut dyn Write) -> Result<()> {
417        let body_offset = write_envelope(writer, self.type_id, self.version)?;
418        let mut entry_writer = CacheEntryWriter::with_pos(writer, body_offset);
419        (self.serialize_body)(value, &mut entry_writer)
420    }
421
422    pub fn supports_row_selection(&self) -> bool {
423        self.row_decoder.is_some()
424    }
425
426    /// Read and validate the envelope of a range-read entry, returning the
427    /// body offset and the entry's `type_version`.
428    fn read_row_envelope(&self, reader: &dyn CacheRangeReader) -> Result<(usize, u32)> {
429        let prefix = reader.read_range(0..ENVELOPE_ID_OFFSET)?;
430        let id_bytes = prefix
431            .get(ENVELOPE_ID_OFFSET - 2..ENVELOPE_ID_OFFSET)
432            .ok_or_else(|| Error::invalid_input("short cache envelope"))?;
433        let id_len = u16::from_le_bytes(id_bytes.try_into().unwrap()) as usize;
434        let header =
435            reader.read_range(0..(ENVELOPE_ID_OFFSET + id_len + ENVELOPE_VERSION_BYTES))?;
436        let envelope = parse_envelope(&header)
437            .ok_or_else(|| Error::invalid_input("invalid cache envelope"))?;
438        if envelope.type_id != self.type_id || envelope.type_version > self.version {
439            return Err(Error::invalid_input("incompatible cache entry"));
440        }
441        Ok((envelope.body_offset, envelope.type_version))
442    }
443
444    /// Gather rows without loading the complete persistent entry.
445    pub fn deserialize_rows(
446        &self,
447        reader: &dyn CacheRangeReader,
448        rows: &[u32],
449    ) -> CacheDecode<ArcAny> {
450        let decode = || -> Result<ArcAny> {
451            let decoder = self
452                .row_decoder
453                .ok_or_else(|| Error::invalid_input("row selection unsupported"))?;
454            let (body_offset, version) = self.read_row_envelope(reader)?;
455            decoder(reader, body_offset, version, rows)
456        };
457        match decode() {
458            Ok(entry) => CacheDecode::Hit(entry),
459            Err(_) => CacheDecode::Miss(CacheMissReason::BodyError),
460        }
461    }
462
463    /// Payload ranges that [`deserialize_rows`](Self::deserialize_rows) would
464    /// request for `rows`, beyond the envelope and body header it reads from
465    /// `reader` here. `None` when the codec cannot plan, the entry is not
466    /// compatible, or the rows are invalid; callers then decode serially,
467    /// which reports invalid rows as a miss.
468    pub fn plan_rows(
469        &self,
470        reader: &dyn CacheRangeReader,
471        rows: &[u32],
472    ) -> Option<Vec<std::ops::Range<usize>>> {
473        let planner = self.row_planner?;
474        let (body_offset, version) = self.read_row_envelope(reader).ok()?;
475        planner(reader, body_offset, version, rows).ok().flatten()
476    }
477
478    /// Deserialize an entry from `data`.
479    ///
480    /// Never fails: any non-fatal failure to interpret the bytes becomes a
481    /// [`CacheDecode::Miss`] with the reason why (see [`CacheMissReason`]).
482    /// Reading from an in-memory [`Bytes`] cannot do I/O, so there is no fault
483    /// channel — a miss is the only non-`Hit` outcome.
484    pub fn deserialize(&self, data: &Bytes) -> CacheDecode<ArcAny> {
485        let Some(envelope) = parse_envelope(data) else {
486            log::debug!("cache entry rejected: missing or invalid envelope");
487            return CacheDecode::Miss(CacheMissReason::InvalidEnvelope);
488        };
489
490        if envelope.type_id != self.type_id {
491            log::debug!(
492                "cache entry type_id mismatch: got {:?}, expected {:?}",
493                envelope.type_id,
494                self.type_id
495            );
496            return CacheDecode::Miss(CacheMissReason::TypeMismatch);
497        }
498
499        // A version newer than this build writes was produced by a newer build
500        // whose body layout we cannot assume to understand. Older/equal versions
501        // are the impl's responsibility to handle (branching on reader.version()).
502        if envelope.type_version > self.version {
503            log::debug!(
504                "cache entry {:?} has unsupported type_version {} (this build writes {})",
505                self.type_id,
506                envelope.type_version,
507                self.version
508            );
509            return CacheDecode::Miss(CacheMissReason::VersionTooNew);
510        }
511
512        let mut reader = CacheEntryReader::new(data, envelope.body_offset, envelope.type_version);
513        match (self.deserialize_body)(&mut reader) {
514            Ok(value) => CacheDecode::Hit(value),
515            Err(e) => {
516                log::debug!(
517                    "cache entry {:?} v{} failed to decode: {e}",
518                    self.type_id,
519                    envelope.type_version
520                );
521                CacheDecode::Miss(CacheMissReason::BodyError)
522            }
523        }
524    }
525}
526
527#[cfg(test)]
528mod tests {
529    use super::*;
530
531    /// A trivial codec used to exercise the envelope and miss semantics
532    /// without pulling in arrow-backed payloads.
533    #[derive(Debug, PartialEq)]
534    struct Widget {
535        n: u32,
536    }
537
538    impl CacheCodecImpl for Widget {
539        const TYPE_ID: &'static str = "test.Widget";
540        const CURRENT_VERSION: u32 = 1;
541
542        fn serialize(&self, writer: &mut CacheEntryWriter<'_>) -> Result<()> {
543            writer.write_raw(&self.n.to_le_bytes())
544        }
545
546        fn deserialize(reader: &mut CacheEntryReader<'_>) -> Result<Self> {
547            let bytes = reader.read_raw()?;
548            let n = u32::from_le_bytes(
549                bytes
550                    .as_ref()
551                    .try_into()
552                    .map_err(|_| Error::io("bad widget".to_string()))?,
553            );
554            Ok(Self { n })
555        }
556    }
557
558    fn serialize_widget(widget: &Widget) -> Bytes {
559        let codec = CacheCodec::from_impl::<Widget>();
560        let any: ArcAny = Arc::new(Widget { n: widget.n });
561        let mut buf = Vec::new();
562        codec.serialize(&any, &mut buf).unwrap();
563        Bytes::from(buf)
564    }
565
566    /// The miss reason, or `None` if the decode was a hit.
567    fn miss_reason(data: &Bytes) -> Option<CacheMissReason> {
568        match deserialize_widget(data) {
569            CacheDecode::Hit(_) => None,
570            CacheDecode::Miss(reason) => Some(reason),
571        }
572    }
573
574    fn deserialize_widget(data: &Bytes) -> CacheDecode<Widget> {
575        let codec = CacheCodec::from_impl::<Widget>();
576        match codec.deserialize(data) {
577            CacheDecode::Hit(any) => {
578                CacheDecode::Hit(Arc::try_unwrap(any.downcast::<Widget>().unwrap()).unwrap())
579            }
580            CacheDecode::Miss(reason) => CacheDecode::Miss(reason),
581        }
582    }
583
584    #[test]
585    fn envelope_roundtrip_hits() {
586        let bytes = serialize_widget(&Widget { n: 0xDEADBEEF });
587        // Sanity: the entry starts with the magic.
588        assert_eq!(&bytes[..4], b"LCE1");
589        let decoded = deserialize_widget(&bytes).hit().unwrap();
590        assert_eq!(decoded, Widget { n: 0xDEADBEEF });
591    }
592
593    #[test]
594    fn has_cache_envelope_detects_magic() {
595        let bytes = serialize_widget(&Widget { n: 1 });
596        assert!(has_cache_envelope(&bytes));
597        assert!(has_cache_envelope(&MAGIC)); // exactly the magic, nothing after
598        assert!(!has_cache_envelope(b"LCE")); // too short
599        assert!(!has_cache_envelope(b"JUNK and more"));
600        assert!(!has_cache_envelope(&[]));
601    }
602
603    #[test]
604    fn wrong_magic_is_miss() {
605        let mut bytes = serialize_widget(&Widget { n: 7 }).to_vec();
606        bytes[0] = b'X';
607        assert_eq!(
608            miss_reason(&Bytes::from(bytes)),
609            Some(CacheMissReason::InvalidEnvelope)
610        );
611    }
612
613    #[test]
614    fn pre_stabilization_blob_is_miss() {
615        // An old unstable blob led with a small u64 LE length prefix (a JSON
616        // header of tens of bytes) — no magic. It must self-heal to a miss.
617        let mut blob = Vec::new();
618        blob.extend_from_slice(&(42u64).to_le_bytes());
619        blob.extend_from_slice(&[0u8; 42]);
620        assert_eq!(
621            miss_reason(&Bytes::from(blob)),
622            Some(CacheMissReason::InvalidEnvelope)
623        );
624
625        // A different unstable shape led with a small u8 tag (0/1/2).
626        assert_eq!(
627            miss_reason(&Bytes::from(vec![0u8, 1, 2, 3])),
628            Some(CacheMissReason::InvalidEnvelope)
629        );
630    }
631
632    #[test]
633    fn unknown_envelope_version_is_miss() {
634        let mut bytes = serialize_widget(&Widget { n: 7 }).to_vec();
635        bytes[4] = 0xFF; // envelope_version byte
636        assert_eq!(
637            miss_reason(&Bytes::from(bytes)),
638            Some(CacheMissReason::InvalidEnvelope)
639        );
640    }
641
642    #[test]
643    fn type_id_mismatch_is_miss() {
644        // Hand-build an envelope with a foreign type_id but valid framing.
645        let mut buf = Vec::new();
646        write_envelope(&mut buf, "some.OtherType", 1).unwrap();
647        buf.extend_from_slice(&(4u64).to_le_bytes());
648        buf.extend_from_slice(&99u32.to_le_bytes());
649        assert_eq!(
650            miss_reason(&Bytes::from(buf)),
651            Some(CacheMissReason::TypeMismatch)
652        );
653    }
654
655    #[test]
656    fn unsupported_future_type_version_is_miss() {
657        // An entry written by a newer build (higher type_version) must miss
658        // rather than be misread by this build.
659        let mut buf = Vec::new();
660        write_envelope(&mut buf, Widget::TYPE_ID, Widget::CURRENT_VERSION + 1).unwrap();
661        lance_arrow::ipc::write_len_prefixed_bytes(&mut buf, &9u32.to_le_bytes()).unwrap();
662        assert_eq!(
663            miss_reason(&Bytes::from(buf)),
664            Some(CacheMissReason::VersionTooNew)
665        );
666    }
667
668    #[test]
669    fn truncated_envelope_is_miss() {
670        let bytes = serialize_widget(&Widget { n: 7 });
671        for cut in [0, 1, 4, 5, 7, 9] {
672            assert_eq!(
673                miss_reason(&bytes.slice(..cut.min(bytes.len()))),
674                Some(CacheMissReason::InvalidEnvelope),
675                "truncating to {cut} bytes should miss as InvalidEnvelope"
676            );
677        }
678    }
679
680    #[test]
681    fn body_decode_error_is_miss() {
682        // Valid envelope, but the body is too short for the widget.
683        let mut buf = Vec::new();
684        write_envelope(&mut buf, Widget::TYPE_ID, Widget::CURRENT_VERSION).unwrap();
685        buf.extend_from_slice(&(1u64).to_le_bytes());
686        buf.push(0u8);
687        assert_eq!(
688            miss_reason(&Bytes::from(buf)),
689            Some(CacheMissReason::BodyError)
690        );
691    }
692
693    #[test]
694    fn plane_tag_is_metadata_only() {
695        let plain = CacheCodec::from_impl::<Widget>();
696        assert_eq!(plain.plane_tag(), None);
697
698        let tagged = plain.with_plane_tag(2).with_memory_priority(1);
699        assert_eq!(tagged.plane_tag(), Some(2));
700        assert_eq!(tagged.memory_priority(), 1);
701        let copied = tagged;
702        assert_eq!(copied.plane_tag(), Some(2));
703
704        // The tag never reaches the wire format.
705        let value: ArcAny = Arc::new(Widget { n: 11 });
706        let (mut plain_bytes, mut tagged_bytes) = (Vec::new(), Vec::new());
707        plain.serialize(&value, &mut plain_bytes).unwrap();
708        tagged.serialize(&value, &mut tagged_bytes).unwrap();
709        assert_eq!(plain_bytes, tagged_bytes);
710        let decoded = tagged
711            .deserialize(&Bytes::from(tagged_bytes))
712            .hit()
713            .unwrap();
714        assert_eq!(decoded.downcast_ref::<Widget>(), Some(&Widget { n: 11 }));
715    }
716
717    /// Row-selectable body that keeps the default (absent) range planner.
718    struct Rows;
719
720    impl CacheCodecImpl for Rows {
721        const TYPE_ID: &'static str = "test.Rows";
722        const CURRENT_VERSION: u32 = 1;
723        const SUPPORTS_ROW_SELECTION: bool = true;
724
725        fn serialize(&self, _writer: &mut CacheEntryWriter<'_>) -> Result<()> {
726            Ok(())
727        }
728
729        fn deserialize(_reader: &mut CacheEntryReader<'_>) -> Result<Self> {
730            Ok(Self)
731        }
732
733        fn deserialize_rows(
734            _reader: &dyn CacheRangeReader,
735            _body_offset: usize,
736            _version: u32,
737            _rows: &[u32],
738        ) -> Result<Self> {
739            Ok(Self)
740        }
741    }
742
743    #[test]
744    fn plan_rows_defaults_to_serial_decode() {
745        let value: ArcAny = Arc::new(Widget { n: 1 });
746        let widget = CacheCodec::from_impl::<Widget>();
747        let mut widget_bytes = Vec::new();
748        widget.serialize(&value, &mut widget_bytes).unwrap();
749        let widget_bytes = Bytes::from(widget_bytes);
750        let read = |range: std::ops::Range<usize>| -> Result<Bytes> {
751            widget_bytes
752                .get(range.clone())
753                .map(|_| widget_bytes.slice(range))
754                .ok_or_else(|| Error::invalid_input("out of range"))
755        };
756        assert!(!widget.supports_row_selection());
757        assert!(widget.plan_rows(&read, &[0]).is_none());
758
759        let rows = CacheCodec::from_impl::<Rows>();
760        let mut rows_bytes = Vec::new();
761        rows.serialize(&(Arc::new(Rows) as ArcAny), &mut rows_bytes)
762            .unwrap();
763        let rows_bytes = Bytes::from(rows_bytes);
764        let read = |range: std::ops::Range<usize>| -> Result<Bytes> {
765            rows_bytes
766                .get(range.clone())
767                .map(|_| rows_bytes.slice(range))
768                .ok_or_else(|| Error::invalid_input("out of range"))
769        };
770        assert!(rows.supports_row_selection());
771        assert!(rows.plan_rows(&read, &[0]).is_none());
772        assert!(matches!(
773            rows.deserialize_rows(&read, &[0]),
774            CacheDecode::Hit(_)
775        ));
776    }
777
778    #[test]
779    fn reader_exposes_envelope_version() {
780        // type_version travels through the envelope to reader.version().
781        let mut buf = Vec::new();
782        write_envelope(&mut buf, Widget::TYPE_ID, 7).unwrap();
783        let body_off = buf.len();
784        // A widget body so the codec can decode it.
785        lance_arrow::ipc::write_len_prefixed_bytes(&mut buf, &5u32.to_le_bytes()).unwrap();
786        let data = Bytes::from(buf);
787
788        let mut r = CacheEntryReader::new(&data, body_off, 7);
789        assert_eq!(r.version(), 7);
790        assert_eq!(r.read_raw().unwrap().as_ref(), 5u32.to_le_bytes());
791    }
792}
lance_core/cache/codec.rs

lance_core/cache/
codec.rs
