performance[vortex-array]: don't call is_valid to count bytes in varbinview

I noticed that the better blocks compressor uses count_referenced_bytes which
calls is_valid on each view and results in an expensive scalar_at call. This
was 30% of our system-wide CPU usage over 12 hours
(see PR for attached screenshot).

This commit iterates over the nullability mask directly to avoid these
expensive calls.

Signed-off-by: Alfonso Subiotto Marques <[email protected]>

Author: asubiotto

vortex-array/src/arrays/varbinview/compact.rs

@@ -60,23 +60,51 @@ impl VarBinViewArray {
     fn count_referenced_bytes(&self) -> u64 {
         match self.validity() {
             Validity::AllInvalid => 0u64,
-            Validity::NonNullable | Validity::AllValid | Validity::Array(_) => self
+            Validity::NonNullable | Validity::AllValid => self
                 .views()
                 .iter()
-                .enumerate()
-                .map(|(idx, &view)| {
-                    if !self.is_valid(idx) || view.is_inlined() {
+                .map(|&view| {
+                    if view.is_inlined() {
                         0u64
                     } else {
                         view.len() as u64
                     }
                 })
                 .sum(),
+            Validity::Array(_) => {
+                let mask = self.validity().to_mask(self.len());
+                match mask.bit_buffer() {
+                    vortex_mask::AllOr::All => self
+                        .views()
+                        .iter()
+                        .map(|&view| {
+                            if view.is_inlined() {
+                                0u64
+                            } else {
+                                view.len() as u64
+                            }
+                        })
+                        .sum(),
+                    vortex_mask::AllOr::None => 0u64,
+                    vortex_mask::AllOr::Some(bit_buffer) => self
+                        .views()
+                        .iter()
+                        .zip(bit_buffer.iter())
+                        .map(|(&view, is_valid)| {
+                            if !is_valid || view.is_inlined() {
+                                0u64
+                            } else {
+                                view.len() as u64
+                            }
+                        })
+                        .sum(),
+                }
+            }
         }
     }
 
     pub(crate) fn buffer_utilizations(&self) -> Vec<BufferUtilization> {
-        let mut utilizations = self
+        let mut utilizations: Vec<BufferUtilization> = self
             .buffers()
             .iter()
             .map(|buf| {
@@ -85,17 +113,41 @@ impl VarBinViewArray {
             })
             .collect();
 
-        if matches!(self.validity(), Validity::AllInvalid) {
-            return utilizations;
-        }
-
-        for (idx, &view) in self.views().iter().enumerate() {
-            if !self.is_valid(idx) || view.is_inlined() {
-                continue;
+        match self.validity() {
+            Validity::AllInvalid => {}
+            Validity::NonNullable | Validity::AllValid => {
+                for &view in self.views().iter() {
+                    if view.is_inlined() {
+                        continue;
+                    }
+                    let view = view.as_view();
+                    utilizations[view.buffer_index as usize].add(view.offset, view.size);
+                }
+            }
+            Validity::Array(_) => {
+                let mask = self.validity().to_mask(self.len());
+                match mask.bit_buffer() {
+                    vortex_mask::AllOr::All => {
+                        for &view in self.views().iter() {
+                            if view.is_inlined() {
+                                continue;
+                            }
+                            let view = view.as_view();
+                            utilizations[view.buffer_index as usize].add(view.offset, view.size);
+                        }
+                    }
+                    vortex_mask::AllOr::None => {}
+                    vortex_mask::AllOr::Some(bit_buffer) => {
+                        for (&view, is_valid) in self.views().iter().zip(bit_buffer.iter()) {
+                            if !is_valid || view.is_inlined() {
+                                continue;
+                            }
+                            let view = view.as_view();
+                            utilizations[view.buffer_index as usize].add(view.offset, view.size);
+                        }
+                    }
+                }
             }
-            let view = view.as_view();
-
-            utilizations[view.buffer_index as usize].add(view.offset, view.size)
         }
 
         utilizations