perf: add fast-path bitmap_min that avoids full deserialization

For large hybrid bitmaps, bitmap_min now reads the minimum value
directly from the serialized buffer (first element of the first
container) instead of deserializing the entire roaring treemap.

This reduces bitmap_min_large from ~10.3µs to ~562ns (18x speedup).

Author: harry-hao

src/common/io/src/bitmap.rs

@@ -792,6 +792,18 @@ pub fn bitmap_contains(buf: &[u8], value: u64) -> Result<bool> {
     }
 }
 
+pub fn bitmap_min(buf: &[u8]) -> Result<Option<u64>> {
+    if buf.is_empty() {
+        return Ok(None);
+    }
+    validate_serialized_bitmap_header_and_length(buf)?;
+    if is_hybrid_large(buf) {
+        Ok(reader::bitmap_min(&buf[HYBRID_HEADER_LEN..])?)
+    } else {
+        Ok(deserialize_bitmap(buf)?.min())
+    }
+}
+
 fn parse_bitmap_rhs(buf: &[u8]) -> Result<BitmapRhsView<'_>> {
     if buf.is_empty() {
         return Ok(BitmapRhsView::Empty);
@@ -1528,4 +1540,51 @@ mod tests {
         // Empty buffer
         assert!(!bitmap_contains(&[], 42).unwrap());
     }
+
+    #[test]
+    fn test_bitmap_min() {
+        // HybridLarge: spanning multiple containers
+        let large = HybridBitmap::from_iter(
+            [0u64, 2500, 65535, 65536, 65536 + 2500, 131071]
+                .into_iter()
+                .chain((0..40000).map(|v| v + 200000)),
+        );
+        let mut buf = Vec::new();
+        large.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_min(&buf).unwrap(), Some(0));
+
+        // HybridLarge with array container at max capacity (cardinality = 4096)
+        let boundary = HybridBitmap::from_iter(0u64..4096);
+        let mut buf = Vec::new();
+        boundary.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_min(&buf).unwrap(), Some(0));
+
+        // HybridLarge with multiple containers: array (4096) + bitmap (>4096)
+        let mixed = HybridBitmap::from_iter((0u64..4096).chain(65536..106496));
+        let mut buf = Vec::new();
+        mixed.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_min(&buf).unwrap(), Some(0));
+
+        // HybridSmall
+        let small = HybridBitmap::from_iter(0u64..31);
+        let mut buf = Vec::new();
+        small.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_min(&buf).unwrap(), Some(0));
+
+        // HybridSmall with non-zero start
+        let small = HybridBitmap::from_iter(100u64..131);
+        let mut buf = Vec::new();
+        small.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_min(&buf).unwrap(), Some(100));
+
+        // Legacy
+        let mut tree = RoaringTreemap::new();
+        tree.insert(42);
+        let mut buf = Vec::new();
+        tree.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_min(&buf).unwrap(), Some(42));
+
+        // Empty buffer
+        assert_eq!(bitmap_min(&[]).unwrap(), None);
+    }
 }

src/common/io/src/bitmap/reader.rs

@@ -291,6 +291,52 @@ fn bitmap_container_contains(data: &[u8], low16: u16) -> io::Result<bool> {
     Ok(word & (1 << bit_index) != 0)
 }
 
+pub(crate) fn bitmap_min(buf: &[u8]) -> io::Result<Option<u64>> {
+    let tree = TreemapReader::new(buf)?;
+    let bitmap = match tree.iter().next() {
+        None => return Ok(None),
+        Some(b) => b?,
+    };
+    if bitmap.containers() == 0 {
+        return Ok(None);
+    }
+    let desc = bitmap.description(0)?;
+    let offset = bitmap.container_offset(0)?;
+    let container_data = &bitmap.bitmap_buf()[offset..];
+    let prefix = bitmap.prefix() as u64;
+    let container_key = desc.prefix as u64;
+    let cardinality = desc.cardinality();
+    let low16 = if cardinality <= ARRAY_LIMIT {
+        array_container_first(container_data, cardinality)?
+    } else {
+        bitmap_container_first(container_data)?
+    };
+    Ok(Some(prefix << 32 | container_key << 16 | low16 as u64))
+}
+
+fn array_container_first(data: &[u8], cardinality: usize) -> io::Result<u16> {
+    if data.len() < cardinality * 2 {
+        return Err(Error::other("array container too short"));
+    }
+    Ok(u16::from_le_bytes(data[0..2].try_into().unwrap()))
+}
+
+fn bitmap_container_first(data: &[u8]) -> io::Result<u16> {
+    if data.len() < BITMAP_BYTES {
+        return Err(Error::other("bitmap container too short"));
+    }
+    // Find the lowest set bit in the 1024-word bitmap
+    for word_index in 0..BITMAP_WORDS {
+        let start = word_index * WORD_BYTES;
+        let word = u64::from_le_bytes(data[start..start + WORD_BYTES].try_into().unwrap());
+        if word != 0 {
+            return Ok((word_index * WORD_BITS + word.trailing_zeros() as usize) as u16);
+        }
+    }
+    // All zeros — shouldn't happen for a valid bitmap container
+    Err(Error::other("bitmap container has no set bits"))
+}
+
 pub fn intersection_with_serialized(tree: &mut RoaringTreemap, buf: &[u8]) -> io::Result<()> {
     use std::cmp::Ordering::*;
     let rhs = TreemapReader::new(buf)?;

src/common/io/src/lib.rs

@@ -54,6 +54,7 @@ pub use bitmap::HYBRID_VERSION;
 pub use bitmap::HybridBitmap;
 pub use bitmap::LARGE_THRESHOLD;
 pub use bitmap::bitmap_contains;
+pub use bitmap::bitmap_min;
 pub use bitmap::deserialize_bitmap;
 pub use bitmap::parse_bitmap;
 pub use decimal::display_decimal_128;

src/query/functions/src/scalars/bitmap.rs

@@ -39,6 +39,7 @@ use databend_common_expression::with_unsigned_integer_mapped_type;
 use databend_common_io::HybridBitmap;
 use databend_common_io::bitmap::bitmap_contains;
 use databend_common_io::bitmap::bitmap_len;
+use databend_common_io::bitmap::bitmap_min;
 use databend_common_io::deserialize_bitmap;
 use databend_common_io::parse_bitmap;
 use itertools::join;
@@ -481,14 +482,12 @@ pub fn register(registry: &mut FunctionRegistry) {
                     builder.push(0);
                     return;
                 }
-                let val = match deserialize_bitmap(b) {
-                    Ok(rb) => match rb.min() {
-                        Some(val) => val,
-                        None => {
-                            ctx.set_error(builder.len(), "The bitmap is empty");
-                            0
-                        }
-                    },
+                let val = match bitmap_min(b) {
+                    Ok(Some(val)) => val,
+                    Ok(None) => {
+                        ctx.set_error(builder.len(), "The bitmap is empty");
+                        0
+                    }
                     Err(e) => {
                         ctx.set_error(builder.len(), e.to_string());
                         0