perf: add fast-path bitmap_max that avoids full deserialization

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

This reduces bitmap_max_large from ~10.7µs to ~531ns (20x speedup).

Author: harry-hao

src/common/io/src/bitmap.rs

@@ -804,6 +804,18 @@ pub fn bitmap_min(buf: &[u8]) -> Result<Option<u64>> {
     }
 }
 
+pub fn bitmap_max(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_max(&buf[HYBRID_HEADER_LEN..])?)
+    } else {
+        Ok(deserialize_bitmap(buf)?.max())
+    }
+}
+
 fn parse_bitmap_rhs(buf: &[u8]) -> Result<BitmapRhsView<'_>> {
     if buf.is_empty() {
         return Ok(BitmapRhsView::Empty);
@@ -1587,4 +1599,52 @@ mod tests {
         // Empty buffer
         assert_eq!(bitmap_min(&[]).unwrap(), None);
     }
+
+    #[test]
+    fn test_bitmap_max() {
+        // 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_max(&buf).unwrap(), Some(239999));
+
+        // 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_max(&buf).unwrap(), Some(4095));
+
+        // 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_max(&buf).unwrap(), Some(106495));
+
+        // HybridSmall
+        let small = HybridBitmap::from_iter(0u64..31);
+        let mut buf = Vec::new();
+        small.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_max(&buf).unwrap(), Some(30));
+
+        // HybridSmall with different range
+        let small = HybridBitmap::from_iter(100u64..131);
+        let mut buf = Vec::new();
+        small.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_max(&buf).unwrap(), Some(130));
+
+        // Legacy
+        let mut tree = RoaringTreemap::new();
+        tree.insert(42);
+        tree.insert(100);
+        let mut buf = Vec::new();
+        tree.serialize_into(&mut buf).unwrap();
+        assert_eq!(bitmap_max(&buf).unwrap(), Some(100));
+
+        // Empty buffer
+        assert_eq!(bitmap_max(&[]).unwrap(), None);
+    }
 }

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

@@ -337,6 +337,61 @@ fn bitmap_container_first(data: &[u8]) -> io::Result<u16> {
     Err(Error::other("bitmap container has no set bits"))
 }
 
+pub(crate) fn bitmap_max(buf: &[u8]) -> io::Result<Option<u64>> {
+    let tree = TreemapReader::new(buf)?;
+    // Find the last prefix bucket
+    let mut last_bitmap = None;
+    for bitmap_result in tree.iter() {
+        last_bitmap = Some(bitmap_result?);
+    }
+    let bitmap = match last_bitmap {
+        None => return Ok(None),
+        Some(b) => b,
+    };
+    if bitmap.containers() == 0 {
+        return Ok(None);
+    }
+    let last_idx = bitmap.containers() - 1;
+    let desc = bitmap.description(last_idx)?;
+    let offset = bitmap.container_offset(last_idx)?;
+    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_last(container_data, cardinality)?
+    } else {
+        bitmap_container_last(container_data)?
+    };
+    Ok(Some(prefix << 32 | container_key << 16 | low16 as u64))
+}
+
+fn array_container_last(data: &[u8], cardinality: usize) -> io::Result<u16> {
+    if data.len() < cardinality * 2 {
+        return Err(Error::other("array container too short"));
+    }
+    let offset = (cardinality - 1) * 2;
+    Ok(u16::from_le_bytes(
+        data[offset..offset + 2].try_into().unwrap(),
+    ))
+}
+
+fn bitmap_container_last(data: &[u8]) -> io::Result<u16> {
+    if data.len() < BITMAP_BYTES {
+        return Err(Error::other("bitmap container too short"));
+    }
+    // Find the highest set bit by scanning from the last word backwards
+    for word_index in (0..BITMAP_WORDS).rev() {
+        let start = word_index * WORD_BYTES;
+        let word = u64::from_le_bytes(data[start..start + WORD_BYTES].try_into().unwrap());
+        if word != 0 {
+            let bit_index = WORD_BITS - 1 - word.leading_zeros() as usize;
+            return Ok((word_index * WORD_BITS + bit_index) as u16);
+        }
+    }
+    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_max;
 pub use bitmap::bitmap_min;
 pub use bitmap::deserialize_bitmap;
 pub use bitmap::parse_bitmap;

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_max;
 use databend_common_io::bitmap::bitmap_min;
 use databend_common_io::deserialize_bitmap;
 use databend_common_io::parse_bitmap;
@@ -450,14 +451,12 @@ pub fn register(registry: &mut FunctionRegistry) {
                     builder.push(0);
                     return;
                 }
-                let val = match deserialize_bitmap(b) {
-                    Ok(rb) => match rb.max() {
-                        Some(val) => val,
-                        None => {
-                            ctx.set_error(builder.len(), "The bitmap is empty");
-                            0
-                        }
-                    },
+                let val = match bitmap_max(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