feat: prefer take to filter for very sparse masks (#1249)

Fixes #1248

It seems to me that if we have fewer than one-in-1024 values there is no
chance for SIMD vectorization to make the mask as fast as a take.

Mild improvement.

```
group                       develop                 layouts                 take-one-in-1024        take-one-in-128         take-one-in-16         always-take
-----                       -------                 -------                 ----------------        ---------------         --------------         -----------
vortex-local-fs             1.00   381.3±10.78µs    34.91    13.3±0.13ms    29.49    11.2±0.11ms    29.63    11.3±0.09ms    29.75    11.3±0.13ms   29.71    11.3±0.12ms
vortex-tokio-local-disk     1.00   349.5±12.64µs    29.46    10.3±0.14ms    24.23     8.5±0.15ms    24.73     8.6±0.09ms    24.80     8.7±0.12ms   24.85     8.7±0.18ms
```

Author: danking

vortex-serde/src/layouts/read/mask.rs

@@ -3,13 +3,15 @@ use std::fmt::{Display, Formatter};
 
 use arrow_buffer::{BooleanBuffer, MutableBuffer};
 use croaring::Bitmap;
-use vortex_array::array::BoolArray;
-use vortex_array::compute::{filter, slice};
+use vortex_array::array::{BoolArray, PrimitiveArray};
+use vortex_array::compute::{filter, slice, take};
 use vortex_array::validity::Validity;
 use vortex_array::{iterate_integer_array, Array, IntoArray};
 use vortex_dtype::PType;
 use vortex_error::{vortex_bail, vortex_err, VortexResult};
 
+const PREFER_TAKE_TO_FILTER_DENSITY: f64 = 1.0 / 1024.0;
+
 /// Bitmap of selected rows within given [begin, end) row range
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct RowMask {
@@ -158,12 +160,20 @@ impl RowMask {
             return Ok(Some(sliced.clone()));
         }
 
-        let predicate = self.to_predicate_array()?;
+        if (true_count as f64 / sliced.len() as f64) < PREFER_TAKE_TO_FILTER_DENSITY {
+            let indices = self.to_indices_array()?;
+            take(sliced, indices).map(Some)
+        } else {
+            let mask = self.to_mask_array()?;
+            filter(sliced, mask).map(Some)
+        }
+    }
 
-        filter(sliced, predicate).map(Some)
+    pub fn to_indices_array(&self) -> VortexResult<Array> {
+        Ok(PrimitiveArray::from_vec(self.values.to_vec(), Validity::NonNullable).into_array())
     }
 
-    pub fn to_predicate_array(&self) -> VortexResult<Array> {
+    pub fn to_mask_array(&self) -> VortexResult<Array> {
         let bitset = self
             .values
             .to_bitset()

vortex-serde/src/layouts/read/stream.rs

@@ -185,7 +185,7 @@ impl<R: VortexReadAt + Unpin + 'static> Stream for LayoutBatchStream<R> {
                                 if let Some(row_mask) = &self.row_mask {
                                     batch = and(
                                         batch,
-                                        row_mask.slice(sel_begin, sel_end).to_predicate_array()?,
+                                        row_mask.slice(sel_begin, sel_end).to_mask_array()?,
                                     )?;
                                 }