Improve intersect_by_rank performance

encodings/fastlanes/src/bitpacking/compute/filter.rs

@@ -226,9 +226,7 @@ mod test {
         let mut ctx = LEGACY_SESSION.create_execution_ctx();
         let unpacked = PrimitiveArray::from_iter((0..4096).map(|i| (i % 63) as u8));
         let bitpacked = BitPackedData::encode(&unpacked.into_array(), 6, &mut ctx).unwrap();
-        let filtered = bitpacked
-            .filter(Mask::from_indices(4096, (0..1024).collect()))
-            .unwrap();
+        let filtered = bitpacked.filter(Mask::from_indices(4096, 0..1024)).unwrap();
         let filtered_prim = filtered.execute::<PrimitiveArray>(&mut ctx).unwrap();
         assert_arrays_eq!(
             filtered_prim,
@@ -243,7 +241,7 @@ mod test {
         let unpacked = PrimitiveArray::new(values.clone(), Validity::NonNullable);
         let bitpacked = BitPackedData::encode(&unpacked.into_array(), 9, &mut ctx).unwrap();
         let filtered = bitpacked
-            .filter(Mask::from_indices(values.len(), (0..250).collect()))
+            .filter(Mask::from_indices(values.len(), 0..250))
             .unwrap()
             .execute::<PrimitiveArray>(&mut ctx)
             .unwrap();

vortex-array/benches/filter_bool.rs

@@ -110,7 +110,7 @@ fn make_dense_runs(len: usize, false_rate: f64, rng: &mut StdRng) -> Mask {
 fn make_single_slice(len: usize, density: f64) -> Mask {
     let true_count = (len as f64 * density) as usize;
     let start = (len - true_count) / 2;
-    Mask::from_indices(len, (start..start + true_count).collect())
+    Mask::from_indices(len, start..start + true_count)
 }
 
 // --- Source array generators ---

vortex-array/src/patches.rs

@@ -1630,7 +1630,7 @@ mod test {
         .unwrap();
 
         // Keep all indices (mask with indices 0-9)
-        let mask = Mask::from_indices(10, (0..10).collect());
+        let mask = Mask::from_indices(10, 0..10);
         let filtered = patches
             .filter(&mask, &mut LEGACY_SESSION.create_execution_ctx())
             .unwrap()

vortex-array/src/scalar_fn/fns/case_when.rs

@@ -1187,8 +1187,8 @@ mod tests {
         let n = 100usize;
 
         // Branch 0: even rows → 0, Branch 1: odd rows → 1, Else: never reached.
-        let branch0_mask = Mask::from_indices(n, (0..n).step_by(2).collect());
-        let branch1_mask = Mask::from_indices(n, (1..n).step_by(2).collect());
+        let branch0_mask = Mask::from_indices(n, (0..n).step_by(2));
+        let branch1_mask = Mask::from_indices(n, (1..n).step_by(2));
 
         let result = merge_case_branches(
             vec![

vortex-array/src/scalar_fn/fns/zip/mod.rs

@@ -432,7 +432,7 @@ mod tests {
             builder.finish()
         };
 
-        let mask = Mask::from_indices(200, (0..100).filter(|i| i % 3 != 0).collect());
+        let mask = Mask::from_indices(200, (0..100).filter(|i| i % 3 != 0));
         let mask_array = mask.clone().into_array();
 
         let mut ctx = LEGACY_SESSION.create_execution_ctx();

vortex-buffer/public-api.lock

@@ -256,7 +256,7 @@ pub fn vortex_buffer::BitBuffer::empty() -> Self
 
 pub fn vortex_buffer::BitBuffer::false_count(&self) -> usize
 
-pub fn vortex_buffer::BitBuffer::from_indices(usize, &[usize]) -> vortex_buffer::BitBuffer
+pub fn vortex_buffer::BitBuffer::from_indices(usize, impl core::iter::traits::collect::IntoIterator<Item = usize>) -> vortex_buffer::BitBuffer
 
 pub fn vortex_buffer::BitBuffer::full(bool, usize) -> Self
 
@@ -452,7 +452,7 @@ pub fn vortex_buffer::BitBufferMut::freeze(self) -> vortex_buffer::BitBuffer
 
 pub fn vortex_buffer::BitBufferMut::from_buffer(vortex_buffer::ByteBufferMut, usize, usize) -> Self
 
-pub fn vortex_buffer::BitBufferMut::from_indices(usize, &[usize]) -> vortex_buffer::BitBufferMut
+pub fn vortex_buffer::BitBufferMut::from_indices(usize, impl core::iter::traits::collect::IntoIterator<Item = usize>) -> vortex_buffer::BitBufferMut
 
 pub fn vortex_buffer::BitBufferMut::full(bool, usize) -> Self
 

vortex-buffer/src/bit/buf.rs

@@ -139,7 +139,7 @@ impl BitBuffer {
     }
 
     /// Create a bit buffer of `len` with `indices` set as true.
-    pub fn from_indices(len: usize, indices: &[usize]) -> BitBuffer {
+    pub fn from_indices(len: usize, indices: impl IntoIterator<Item = usize>) -> BitBuffer {
         BitBufferMut::from_indices(len, indices).freeze()
     }
 
@@ -650,6 +650,24 @@ mod tests {
         assert_eq!(sliced.offset(), 2);
     }
 
+    #[test]
+    fn test_from_indices_dense_crosses_words() {
+        let len = 130;
+        let indices = (0..len).filter(|idx| idx % 3 != 1);
+        let buf = BitBuffer::from_indices(len, indices);
+
+        assert_eq!(buf.len(), len);
+        for idx in 0..len {
+            assert_eq!(buf.value(idx), idx % 3 != 1, "mismatch at {idx}");
+        }
+    }
+
+    #[test]
+    #[should_panic(expected = "index 5 exceeds len 5")]
+    fn test_from_indices_out_of_bounds() {
+        BitBuffer::from_indices(5, [0, 5]);
+    }
+
     #[rstest]
     #[case(5)]
     #[case(8)]

vortex-buffer/src/bit/buf_mut.rs

@@ -165,11 +165,21 @@ impl BitBufferMut {
     }
 
     /// Create a bit buffer of `len` with `indices` set as true.
-    pub fn from_indices(len: usize, indices: &[usize]) -> BitBufferMut {
-        let mut buf = BitBufferMut::new_unset(len);
-        // TODO(ngates): for dense indices, we can do better by collecting into u64s.
-        indices.iter().for_each(|&idx| buf.set(idx));
-        buf
+    pub fn from_indices(len: usize, indices: impl IntoIterator<Item = usize>) -> BitBufferMut {
+        let mut buffer = BufferMut::<u64>::zeroed(len.div_ceil(64));
+        for idx in indices {
+            assert!(idx < len, "index {idx} exceeds len {len}");
+            buffer.as_mut_slice()[idx / 64] |= 1 << (idx % 64);
+        }
+
+        let mut buffer = buffer.into_byte_buffer();
+        buffer.truncate(len.div_ceil(8));
+
+        Self {
+            buffer,
+            offset: 0,
+            len,
+        }
     }
 
     /// Invokes `f` with indexes `0..len` collecting the boolean results into a new `BitBufferMut`

vortex-mask/benches/intersect_by_rank.rs

@@ -68,12 +68,7 @@ fn create_random_mask(len: usize, selectivity: f64) -> Mask {
 fn create_random_indices_mask(len: usize, selectivity: f64) -> Mask {
     #[expect(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
     let threshold = (selectivity * 1000.0) as usize;
-    Mask::from_indices(
-        len,
-        (0..len)
-            .filter(|&i| (i * 7 + 13) % 1000 < threshold)
-            .collect(),
-    )
+    Mask::from_indices(len, (0..len).filter(|&i| (i * 7 + 13) % 1000 < threshold))
 }
 
 fn create_runs_mask(len: usize, run_len: usize, gap_len: usize) -> Mask {

vortex-mask/public-api.lock

@@ -56,7 +56,7 @@ pub fn vortex_mask::Mask::from_buffer(vortex_buffer::bit::buf::BitBuffer) -> Sel
 
 pub fn vortex_mask::Mask::from_excluded_indices(usize, impl core::iter::traits::collect::IntoIterator<Item = usize>) -> Self
 
-pub fn vortex_mask::Mask::from_indices(usize, alloc::vec::Vec<usize>) -> Self
+pub fn vortex_mask::Mask::from_indices(usize, impl core::iter::traits::collect::IntoIterator<Item = usize>) -> Self
 
 pub fn vortex_mask::Mask::from_intersection_indices(usize, impl core::iter::traits::iterator::Iterator<Item = usize>, impl core::iter::traits::iterator::Iterator<Item = usize>) -> Self