refactor(rust): carve v1 wire-format seams to prep for v2 (#1513)

Prepare for v2 with no wire-format or behavior change

- Convert RawPresence from a tuple struct over Option<RawStream> into an
enum (AllPresent / Stream) and centralize all presence decoding behind
RawPresence::decode_bits / decode_bools. v2 can then add raw-bitfield
and shared-presence-group variants without touching downstream code.
- Move the v1 stream-header wire codec (stream_type byte, encoding byte,
counts, RLE/Morton extras) into decoder/stream/header01.rs, absorbing
the stream_type byte codec from physical.rs. StreamMeta / RawStream /
IntEncoding constructors move to model.rs as format-independent
in-memory types, so a future header02 codec can sit alongside.
- Rename parse_struct_children to parse_shared_dict_children to match
shared-dictionary terminology used elsewhere (per #1201).

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: nyurik

rust/mlt-core/src/decoder/analyze.rs

@@ -78,7 +78,10 @@ impl Analyze for RawIdValue<'_> {
 
 impl Analyze for RawPresence<'_> {
     fn for_each_stream(&self, cb: &mut dyn FnMut(StreamMeta)) {
-        self.0.for_each_stream(cb);
+        match self {
+            Self::AllPresent => {}
+            Self::Stream(s) => s.for_each_stream(cb),
+        }
     }
 }
 

rust/mlt-core/src/decoder/property/decode.rs

@@ -1,7 +1,36 @@
+use std::borrow::Cow;
+
+use bitvec::order::Lsb0;
+use bitvec::slice::BitSlice;
+
 use crate::decoder::{ParsedProperty, ParsedScalar, RawPresence, RawProperty};
 use crate::utils::decode_presence;
 use crate::{Decode, Decoder, MltResult};
 
+impl<'a> RawPresence<'a> {
+    /// Decode into a packed bitvector, or `None` for a non-optional column.
+    ///
+    /// This is the only place aware of every wire representation of presence;
+    /// all downstream presence handling goes through the returned bits.
+    pub(crate) fn decode_bits(
+        self,
+        dec: &mut Decoder,
+    ) -> MltResult<Option<Cow<'a, BitSlice<u8, Lsb0>>>> {
+        match self {
+            Self::AllPresent => Ok(None),
+            Self::Stream(s) => Ok(Some(s.decode_bitvec(dec)?)),
+        }
+    }
+
+    /// Decode into one bool per feature, or `None` for a non-optional column.
+    pub(crate) fn decode_bools(self, dec: &mut Decoder) -> MltResult<Option<Vec<bool>>> {
+        match self {
+            Self::AllPresent => Ok(None),
+            Self::Stream(s) => Ok(Some(s.decode_bools(dec)?)),
+        }
+    }
+}
+
 impl<'a, T: Copy + PartialEq> ParsedScalar<'a, T> {
     pub fn from_parts(
         name: &'a str,

rust/mlt-core/src/decoder/property/model.rs

@@ -209,6 +209,16 @@ pub struct RawFsstData<'a> {
     pub corpus: RawStream<'a>,
 }
 
-/// Raw presence/nullability stream borrowed from input bytes.
+/// Raw presence/nullability data for a column.
+///
+/// `AllPresent` represents a non-optional column; other variants encode presence in a
+/// layer-format-specific way. Decode through [`RawPresence::decode_bits`] or
+/// [`RawPresence::decode_bools`] instead of matching on it.
 #[derive(Debug, Clone, PartialEq, Default)]
-pub struct RawPresence<'a>(pub Option<RawStream<'a>>);
+pub enum RawPresence<'a> {
+    /// Non-optional column — every feature has a value; nothing is stored.
+    #[default]
+    AllPresent,
+    /// Tag `0x01`: bool-RLE presence stream with a full stream header.
+    Stream(RawStream<'a>),
+}

rust/mlt-core/src/decoder/property/strings.rs

@@ -280,10 +280,7 @@ impl<'a> RawStrings<'a> {
     /// Decode string property from its encoded column.
     pub fn decode(self, dec: &mut Decoder) -> MltResult<ParsedStrings<'a>> {
         let name = self.name;
-        let presence = match self.presence.0 {
-            Some(s) => Some(s.decode_bools(dec)?),
-            None => None,
-        };
+        let presence = self.presence.decode_bools(dec)?;
 
         let parsed = match self.encoding {
             RawStringsEncoding::Plain(plain_data) => {
@@ -448,10 +445,7 @@ impl<'a> RawSharedDict<'a> {
         let mut items = Vec::with_capacity(self.children.len());
         for child in self.children {
             let offsets: Vec<u32> = child.data.decode_u32s(dec)?;
-            let presence = match child.presence.0 {
-                Some(s) => Some(s.decode_bools(dec)?),
-                None => None,
-            };
+            let presence = child.presence.decode_bools(dec)?;
             let ranges = resolve_dict_spans(&offsets, presence.as_deref(), &dict_spans, dec)?
                 .into_iter()
                 .map(|span| match span {

rust/mlt-core/src/decoder/root.rs

@@ -291,74 +291,74 @@ impl<'a> Layer01<'a, Lazy> {
         for column in col_info {
             use crate::decoder::RawProperty as RP;
 
-            let opt;
+            let presence;
             let value;
             let name = column.name.unwrap_or("");
 
             match column.typ {
                 ColumnType::Id | ColumnType::OptId => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
                     id_column.set_once(Raw(RawId {
-                        presence: RawPresence(opt),
+                        presence,
                         value: RawIdValue::Id32(value),
                     }))?;
                 }
                 ColumnType::LongId | ColumnType::OptLongId => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
                     id_column.set_once(Raw(RawId {
-                        presence: RawPresence(opt),
+                        presence,
                         value: RawIdValue::Id64(value),
                     }))?;
                 }
                 ColumnType::Geometry => {
                     input = parse_geometry_column(input, &mut geometry, parser)?;
                 }
                 ColumnType::Bool | ColumnType::OptBool => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::parse_bool(input, parser)?;
-                    properties.push(Raw(RP::Bool(scalar(name, opt, value))));
+                    properties.push(Raw(RP::Bool(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::I8 | ColumnType::OptI8 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::I8(scalar(name, opt, value))));
+                    properties.push(Raw(RP::I8(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::U8 | ColumnType::OptU8 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::U8(scalar(name, opt, value))));
+                    properties.push(Raw(RP::U8(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::I32 | ColumnType::OptI32 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::I32(scalar(name, opt, value))));
+                    properties.push(Raw(RP::I32(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::U32 | ColumnType::OptU32 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::U32(scalar(name, opt, value))));
+                    properties.push(Raw(RP::U32(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::I64 | ColumnType::OptI64 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::I64(scalar(name, opt, value))));
+                    properties.push(Raw(RP::I64(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::U64 | ColumnType::OptU64 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::U64(scalar(name, opt, value))));
+                    properties.push(Raw(RP::U64(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::F32 | ColumnType::OptF32 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::F32(scalar(name, opt, value))));
+                    properties.push(Raw(RP::F32(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::F64 | ColumnType::OptF64 => {
-                    (input, opt) = parse_optional(column.typ, input, parser)?;
+                    (input, presence) = parse_optional(column.typ, input, parser)?;
                     (input, value) = RawStream::from_bytes(input, parser)?;
-                    properties.push(Raw(RP::F64(scalar(name, opt, value))));
+                    properties.push(Raw(RP::F64(RawScalar::new(name, presence, value))));
                 }
                 ColumnType::Str | ColumnType::OptStr => {
                     let prop;
@@ -408,26 +408,26 @@ impl<'a> Layer01<'a, Lazy> {
     }
 }
 
-fn parse_struct_children<'a>(
+fn parse_shared_dict_children<'a>(
     mut input: &'a [u8],
     column: &Column<'a>,
     parser: &mut Parser,
 ) -> MltRefResult<'a, Vec<RawSharedDictItem<'a>>> {
     let mut children = Vec::with_capacity(column.children.len());
     for child in &column.children {
         let (inp, sc) = parse_varint::<u32>(input)?;
-        let (inp, child_optional) = parse_optional(child.typ, inp, parser)?;
-        let optional_stream_count = u32::from(child_optional.is_some());
+        let (inp, presence) = parse_optional(child.typ, inp, parser)?;
+        let optional_stream_count = u32::from(presence.is_optional());
         if let Some(data_count) = sc.checked_sub(optional_stream_count)
             && data_count != 1
         {
             return Err(UnexpectedStructChildCount(data_count));
         }
-        let (inp, child_data) = RawStream::from_bytes(inp, parser)?;
+        let (inp, data) = RawStream::from_bytes(inp, parser)?;
         children.push(RawSharedDictItem {
             name: child.name.unwrap_or(""),
-            presence: RawPresence(child_optional),
-            data: child_data,
+            presence,
+            data,
         });
         input = inp;
     }
@@ -438,12 +438,12 @@ fn parse_optional<'a>(
     typ: ColumnType,
     input: &'a [u8],
     parser: &mut Parser,
-) -> MltRefResult<'a, Option<RawStream<'a>>> {
+) -> MltRefResult<'a, RawPresence<'a>> {
     if typ.is_optional() {
         let (input, optional) = RawStream::parse_bool(input, parser)?;
-        Ok((input, Some(optional)))
+        Ok((input, RawPresence::Stream(optional)))
     } else {
-        Ok((input, None))
+        Ok((input, RawPresence::AllPresent))
     }
 }
 
@@ -482,7 +482,7 @@ fn parse_str_column<'a>(
     };
     let presence;
     (input, presence) = parse_optional(typ, input, parser)?;
-    if presence.is_some() {
+    if presence.is_optional() {
         if stream_count == 0 {
             return Err(UnsupportedStringStreamCount(stream_count));
         }
@@ -516,7 +516,7 @@ fn parse_str_column<'a>(
         input,
         RawProperty::Str(RawStrings {
             name,
-            presence: RawPresence(presence),
+            presence,
             encoding,
         }),
     ))
@@ -548,13 +548,13 @@ fn parse_shared_dict_column<'a>(
         }
     }
     let children;
-    (input, children) = parse_struct_children(input, column, parser)?;
+    (input, children) = parse_shared_dict_children(input, column, parser)?;
 
     // Validate stream_count: must equal dict_streams + children + optional_children.
     let children_n = u32::try_from(children.len()).or_overflow()?;
     let optional_n = children
         .iter()
-        .filter(|c| c.presence.0.is_some())
+        .filter(|c| c.presence.is_optional())
         .count()
         .try_into()
         .or_overflow()?;
@@ -638,11 +638,21 @@ fn parse_columns_meta<'a>(
     Ok((input, (col_info, column_count - geometries - ids)))
 }
 
-fn scalar<'a>(name: &'a str, opt: Option<RawStream<'a>>, value: RawStream<'a>) -> RawScalar<'a> {
-    RawScalar {
-        name,
-        presence: RawPresence(opt),
-        data: value,
+impl<'a> RawScalar<'a> {
+    fn new(name: &'a str, presence: RawPresence<'a>, data: RawStream<'a>) -> Self {
+        Self {
+            name,
+            presence,
+            data,
+        }
+    }
+}
+
+impl RawPresence<'_> {
+    /// Whether this column carries presence data (some features may be null).
+    #[must_use]
+    pub(crate) fn is_optional(&self) -> bool {
+        !matches!(self, Self::AllPresent)
     }
 }