feat: support list blob scans

rust/lance-core/src/datatypes/field.rs

@@ -549,6 +549,20 @@ impl Field {
             .get(ARROW_EXT_NAME_KEY)
             .map(|name| name == BLOB_V2_EXT_NAME)
             .unwrap_or(false)
+            || self.is_blob_v2_descriptor()
+    }
+
+    fn is_blob_v2_descriptor(&self) -> bool {
+        self.metadata.contains_key(BLOB_META_KEY)
+            && self.logical_type == BLOB_V2_DESC_LANCE_FIELD.logical_type
+            && self.children.len() == BLOB_V2_DESC_LANCE_FIELD.children.len()
+            && self
+                .children
+                .iter()
+                .zip(BLOB_V2_DESC_LANCE_FIELD.children.iter())
+                .all(|(child, expected)| {
+                    child.name == expected.name && child.data_type() == expected.data_type()
+                })
     }
 
     // Blob columns intentionally have two schema representations:
@@ -575,6 +589,29 @@ impl Field {
         }
     }
 
+    /// Convert a blob field to the materialized binary payload view.
+    ///
+    /// The field keeps its name and id but uses `LargeBinary` with no children.
+    /// Blob v2 fields retain their extension marker internally so scan planning
+    /// can recognize the binary view before exposing a plain Arrow binary field.
+    pub fn binary_blob_mut(&mut self) {
+        if !self.is_blob() {
+            return;
+        }
+        let is_blob_v2 = self.is_blob_v2();
+
+        self.logical_type = LogicalType::try_from(&DataType::LargeBinary).unwrap();
+        self.children.clear();
+        self.encoding = Some(Encoding::VarBinary);
+        if is_blob_v2 {
+            self.metadata.remove(BLOB_META_KEY);
+            self.metadata.remove("packed");
+            self.metadata.remove("lance-encoding:packed");
+            self.metadata
+                .insert(ARROW_EXT_NAME_KEY.to_string(), BLOB_V2_EXT_NAME.to_string());
+        }
+    }
+
     /// Convert blob v2 fields in this field tree to their descriptor view.
     pub fn unload_blobs_recursive(&mut self) {
         if self.is_blob_v2() {
@@ -812,6 +849,13 @@ impl Field {
         }
 
         if self.is_blob() != other.is_blob() {
+            if ignore_types {
+                return Ok(if self.id >= 0 {
+                    self.clone()
+                } else {
+                    other.clone()
+                });
+            }
             return Err(Error::arrow(format!(
                 "Attempt to intersect blob and non-blob field: {}",
                 self.name
@@ -847,7 +891,7 @@ impl Field {
                 .iter()
                 .filter_map(|c| {
                     if let Some(other_child) = other.child(&c.name) {
-                        let intersection = c.intersection(other_child).ok()?;
+                        let intersection = c.do_intersection(other_child, ignore_types).ok()?;
                         Some(intersection)
                     } else {
                         None
@@ -1847,13 +1891,27 @@ mod tests {
     #[test]
     fn blob_unloaded_mut_selects_layout_from_metadata() {
         let metadata = HashMap::from([(BLOB_META_KEY.to_string(), "true".to_string())]);
+        let mut binary_field: Field = ArrowField::new("blob", DataType::LargeBinary, true)
+            .with_metadata(metadata.clone())
+            .try_into()
+            .unwrap();
+        binary_field.binary_blob_mut();
+        assert!(binary_field.metadata.contains_key(BLOB_META_KEY));
+        assert!(!binary_field.is_blob_v2());
+
         let mut field: Field = ArrowField::new("blob", DataType::LargeBinary, true)
             .with_metadata(metadata)
             .try_into()
             .unwrap();
         field.unloaded_mut();
         assert_eq!(field.children.len(), 2);
         assert_eq!(field.logical_type, BLOB_DESC_LANCE_FIELD.logical_type);
+        assert!(field.is_blob());
+        assert!(!field.is_blob_v2());
+        field.unloaded_mut();
+        assert_eq!(field.children.len(), 2);
+        assert_eq!(field.logical_type, BLOB_DESC_LANCE_FIELD.logical_type);
+        assert!(!field.is_blob_v2());
 
         let metadata =
             HashMap::from([(ARROW_EXT_NAME_KEY.to_string(), BLOB_V2_EXT_NAME.to_string())]);
@@ -1874,6 +1932,12 @@ mod tests {
         field.unloaded_mut();
         assert_eq!(field.children.len(), 5);
         assert_eq!(field.logical_type, BLOB_V2_DESC_LANCE_FIELD.logical_type);
+        assert!(!field.metadata.contains_key(ARROW_EXT_NAME_KEY));
+        assert!(field.is_blob_v2());
+        field.unloaded_mut();
+        assert_eq!(field.children.len(), 5);
+        assert_eq!(field.logical_type, BLOB_V2_DESC_LANCE_FIELD.logical_type);
+        assert!(!field.metadata.contains_key(ARROW_EXT_NAME_KEY));
     }
 
     #[test]

rust/lance-core/src/datatypes/schema.rs

@@ -1071,6 +1071,17 @@ pub enum BlobHandling {
 }
 
 impl BlobHandling {
+    fn should_load_binary(&self, field: &Field) -> bool {
+        if !field.is_blob() {
+            return false;
+        }
+        match self {
+            Self::AllBinary => true,
+            Self::SomeBlobsBinary(set) | Self::SomeBinary(set) => set.contains(&(field.id as u32)),
+            Self::BlobsDescriptions | Self::AllDescriptions => false,
+        }
+    }
+
     fn should_unload(&self, field: &Field) -> bool {
         // Blob v2 columns are Structs, so we need to treat any blob-marked field as unloadable
         // even if the physical data type is not binary-like.
@@ -1086,10 +1097,25 @@ impl BlobHandling {
         }
     }
 
+    /// Apply this blob handling policy to a projected field tree.
+    ///
+    /// Blob descriptor modes convert blob leaves to descriptor views. Binary
+    /// modes convert selected blob leaves to `LargeBinary`. Non-blob nested
+    /// fields are preserved while their children are handled recursively.
     pub fn unload_if_needed(&self, mut field: Field) -> Field {
+        if self.should_load_binary(&field) {
+            field.binary_blob_mut();
+            return field;
+        }
         if self.should_unload(&field) {
             field.unloaded_mut();
+            return field;
         }
+        field.children = field
+            .children
+            .into_iter()
+            .map(|child| self.unload_if_needed(child))
+            .collect();
         field
     }
 }

rust/lance-encoding/src/encodings/logical/blob.rs

@@ -267,16 +267,11 @@ impl FieldEncoder for BlobV2StructuralEncoder {
         &mut self,
         array: ArrayRef,
         external_buffers: &mut OutOfLineBuffers,
-        mut repdef: RepDefBuilder,
+        repdef: RepDefBuilder,
         row_number: u64,
         num_rows: u64,
     ) -> Result<Vec<EncodeTask>> {
         let struct_arr = array.as_struct();
-        if let Some(validity) = struct_arr.nulls() {
-            repdef.add_validity_bitmap(validity.clone());
-        } else {
-            repdef.add_no_null(struct_arr.len());
-        }
 
         let kind_col = struct_arr
             .column_by_name("kind")
@@ -403,7 +398,7 @@ impl FieldEncoder for BlobV2StructuralEncoder {
         let descriptor_array = Arc::new(StructArray::try_new(
             BLOB_V2_DESC_FIELDS.clone(),
             children,
-            None,
+            struct_arr.nulls().cloned(),
         )?) as ArrayRef;
 
         self.descriptor_encoder.maybe_encode(