From 3e1308bc381929818f81d877cd4d16d47682dbf4 Mon Sep 17 00:00:00 2001
From: Nicholas Gates <nick@nickgates.com>
Date: Mon, 13 Oct 2025 15:08:37 -0400
Subject: [PATCH 1/3] Port bit-buffer over

Signed-off-by: Nicholas Gates <nick@nickgates.com>
---
 Cargo.lock                         |   1 +
 vortex-buffer/Cargo.toml           |   1 +
 vortex-buffer/src/bit/aligned.rs   | 127 ++++++++
 vortex-buffer/src/bit/arrow.rs     |  62 ++++
 vortex-buffer/src/bit/buf.rs       | 461 +++++++++++++++++++++++++++++
 vortex-buffer/src/bit/buf_mut.rs   | 351 ++++++++++++++++++++++
 vortex-buffer/src/bit/mod.rs       |  35 +++
 vortex-buffer/src/bit/ops.rs       |  95 ++++++
 vortex-buffer/src/bit/unaligned.rs | 351 ++++++++++++++++++++++
 9 files changed, 1484 insertions(+)
 create mode 100644 vortex-buffer/src/bit/aligned.rs
 create mode 100644 vortex-buffer/src/bit/arrow.rs
 create mode 100644 vortex-buffer/src/bit/buf.rs
 create mode 100644 vortex-buffer/src/bit/buf_mut.rs
 create mode 100644 vortex-buffer/src/bit/mod.rs
 create mode 100644 vortex-buffer/src/bit/ops.rs
 create mode 100644 vortex-buffer/src/bit/unaligned.rs

diff --git a/Cargo.lock b/Cargo.lock
index 44be528168f..52aaaebefa1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8594,6 +8594,7 @@ name = "vortex-buffer"
 version = "0.1.0"
 dependencies = [
  "arrow-buffer",
+ "bitvec",
  "bytes",
  "codspeed-divan-compat",
  "cudarc",
diff --git a/vortex-buffer/Cargo.toml b/vortex-buffer/Cargo.toml
index 6233ad894bb..c81972d4ba7 100644
--- a/vortex-buffer/Cargo.toml
+++ b/vortex-buffer/Cargo.toml
@@ -25,6 +25,7 @@ warn-copy = ["dep:log"]
 
 [dependencies]
 arrow-buffer = { workspace = true, optional = true }
+bitvec = { workspace = true }
 bytes = { workspace = true }
 cudarc = { workspace = true, optional = true }
 itertools = { workspace = true }
diff --git a/vortex-buffer/src/bit/aligned.rs b/vortex-buffer/src/bit/aligned.rs
new file mode 100644
index 00000000000..9e083181aed
--- /dev/null
+++ b/vortex-buffer/src/bit/aligned.rs
@@ -0,0 +1,127 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::iter;
+
+use vortex_error::VortexExpect;
+
+use crate::ByteBuffer;
+use crate::trusted_len::TrustedLen;
+
+/// Aligned bitwise view over underlying bytes in u64 chunks
+pub struct BitChunks {
+    buffer: ByteBuffer,
+    len: usize,
+    bit_offset: usize,
+    remainder_len: usize,
+}
+
+impl BitChunks {
+    /// Construct new with given length and offset
+    pub fn new(buffer: ByteBuffer, offset: usize, len: usize) -> Self {
+        let byte_len = (offset + len).div_ceil(8);
+        assert!(
+            byte_len <= buffer.len(),
+            "Buffer {} too small for given length {len} and offset {offset}",
+            buffer.len()
+        );
+
+        let bit_offset = offset % 8;
+        let byte_offset = offset / 8;
+        let remainder_len = len % 64;
+
+        Self {
+            buffer: buffer.slice(byte_offset..byte_len),
+            len,
+            bit_offset,
+            remainder_len,
+        }
+    }
+
+    /// Length of the last non full slice of bits
+    pub fn remainder_len(&self) -> usize {
+        self.remainder_len
+    }
+
+    /// Last u64 chunk of the underlying buffer
+    pub fn remainder_bits(&self) -> u64 {
+        if self.remainder_len == 0 {
+            return 0;
+        }
+
+        // Since we sliced the buffer on construction then remainder is aligned with the buffer
+        // NOTE: you want the rounding behaviour of integer division i.e., it's not correct to simplify this to self.len / 8
+        let remainder_bytes = &self.buffer[self.len / 64 * 8..];
+        let mut result_bits = remainder_bytes[0] as u64 >> self.bit_offset;
+        for (i, &byte) in remainder_bytes[1..].iter().enumerate() {
+            result_bits |= (byte as u64) << ((i + 1) * 8 - self.bit_offset);
+        }
+
+        result_bits & ((1 << self.remainder_len) - 1)
+    }
+
+    /// Get an interator over the bitwise chunks including the trailer
+    pub fn iter(&self) -> PaddedBitChunksIterator {
+        BitChunksIterator {
+            buffer: self.buffer.clone(),
+            bit_offset: self.bit_offset,
+            chunk_count: self.len / 64,
+            index: 0,
+        }
+        .chain(iter::once(self.remainder_bits()))
+    }
+}
+
+impl IntoIterator for BitChunks {
+    type Item = u64;
+    type IntoIter = PaddedBitChunksIterator;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.iter()
+    }
+}
+
+pub type PaddedBitChunksIterator = iter::Chain<BitChunksIterator, iter::Once<u64>>;
+
+unsafe impl TrustedLen for PaddedBitChunksIterator {}
+
+pub struct BitChunksIterator {
+    buffer: ByteBuffer,
+    bit_offset: usize,
+    chunk_count: usize,
+    index: usize,
+}
+
+impl Iterator for BitChunksIterator {
+    type Item = u64;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.index >= self.chunk_count {
+            return None;
+        }
+
+        let non_offset_chunk = u64::from_le_bytes(
+            self.buffer[self.index * 8..(self.index + 1) * 8]
+                .try_into()
+                .vortex_expect("slice of 8 bytes"),
+        );
+        let result = if self.bit_offset == 0 {
+            non_offset_chunk
+        } else {
+            let next_byte = self.buffer[(self.index + 1) * 8] as u64;
+            (non_offset_chunk >> self.bit_offset) | (next_byte << (64 - self.bit_offset))
+        };
+
+        self.index += 1;
+        Some(result)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let size = self.chunk_count - self.index;
+        (size, Some(size))
+    }
+}
+
+impl ExactSizeIterator for BitChunksIterator {}
+
+unsafe impl TrustedLen for BitChunksIterator {}
diff --git a/vortex-buffer/src/bit/arrow.rs b/vortex-buffer/src/bit/arrow.rs
new file mode 100644
index 00000000000..5c62b33f776
--- /dev/null
+++ b/vortex-buffer/src/bit/arrow.rs
@@ -0,0 +1,62 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Conversions between `BitBuffer` and Arrow's `BooleanBuffer`.
+
+use arrow_buffer::BooleanBuffer;
+
+use crate::{Alignment, BitBuffer, ByteBuffer};
+
+impl From<BooleanBuffer> for BitBuffer {
+    fn from(value: BooleanBuffer) -> Self {
+        let offset = value.offset();
+        let len = value.len();
+        let buffer = value.into_inner();
+        let buffer = ByteBuffer::from_arrow_buffer(buffer, Alignment::of::<u8>());
+
+        BitBuffer::new_with_offset(buffer, len, offset)
+    }
+}
+
+impl From<BitBuffer> for BooleanBuffer {
+    fn from(value: BitBuffer) -> Self {
+        let offset = value.offset();
+        let len = value.len();
+        let buffer = value.into_inner();
+
+        BooleanBuffer::new(buffer.into_arrow_buffer(), offset, len)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder};
+
+    use crate::BitBuffer;
+
+    #[test]
+    fn test_from_arrow() {
+        let mut arrow_bools = BooleanBufferBuilder::new(10);
+        arrow_bools.append_n(5, true);
+        arrow_bools.append_n(5, false);
+        let bit_buffer: BitBuffer = arrow_bools.finish().into();
+
+        for i in 0..5 {
+            assert!(bit_buffer.value(i));
+        }
+
+        for i in 5..10 {
+            assert!(!bit_buffer.value(i));
+        }
+
+        // Convert back to Arrow
+        let arrow_bools: BooleanBuffer = bit_buffer.into();
+
+        for i in 0..5 {
+            assert!(arrow_bools.value(i));
+        }
+        for i in 5..10 {
+            assert!(!arrow_bools.value(i));
+        }
+    }
+}
diff --git a/vortex-buffer/src/bit/buf.rs b/vortex-buffer/src/bit/buf.rs
new file mode 100644
index 00000000000..21e1a081508
--- /dev/null
+++ b/vortex-buffer/src/bit/buf.rs
@@ -0,0 +1,461 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::ops::{BitAnd, BitOr, BitXor, Not, Range};
+
+use crate::bit::aligned::BitChunks;
+use crate::bit::ops::{bitwise_and, bitwise_not, bitwise_or, bitwise_unary_op, bitwise_xor};
+use crate::bit::unaligned::{BitIndexIterator, BitIterator, BitSliceIterator, UnalignedBitChunks};
+use crate::{
+    Alignment, BitBufferMut, Buffer, BufferMut, ByteBuffer, buffer, get_bit, get_bit_unchecked,
+};
+
+/// An immutable bitset stored as a packed byte buffer.
+#[derive(Clone, Debug, Eq)]
+pub struct BitBuffer {
+    buffer: ByteBuffer,
+    len: usize,
+    offset: usize,
+}
+
+impl PartialEq for BitBuffer {
+    fn eq(&self, other: &Self) -> bool {
+        if self.len != other.len {
+            return false;
+        }
+
+        self.chunks()
+            .iter()
+            .zip(other.chunks())
+            .all(|(a, b)| a == b)
+    }
+}
+
+impl BitBuffer {
+    /// Create a new `BoolBuffer` backed by a [`ByteBuffer`] with `len` bits in view.
+    ///
+    /// Panics if the buffer is not large enough to hold `len` bits.
+    pub fn new(buffer: ByteBuffer, len: usize) -> Self {
+        assert!(
+            buffer.len() * 8 >= len,
+            "provided ByteBuffer not large enough to back BoolBuffer with len {len}"
+        );
+
+        Self {
+            buffer,
+            len,
+            offset: 0,
+        }
+    }
+
+    /// Create a new `BoolBuffer` backed by a [`ByteBuffer`] with `len` bits in view, starting at the
+    /// given `offset` (in bits).
+    ///
+    /// Panics if the buffer is not large enough to hold `len` bits or if the offset is greater than
+    pub fn new_with_offset(buffer: ByteBuffer, len: usize, offset: usize) -> Self {
+        assert!(
+            len.saturating_add(offset) <= buffer.len().saturating_mul(8),
+            "provided ByteBuffer (len={}) not large enough to back BoolBuffer with offset {offset} len {len}",
+            buffer.len()
+        );
+
+        Self {
+            buffer,
+            len,
+            offset,
+        }
+    }
+
+    /// Create a new `BoolBuffer` of length `len` where all bits are set (true).
+    pub fn new_set(len: usize) -> Self {
+        let words = len.div_ceil(8);
+        let buffer = buffer![0xFF; words];
+
+        Self {
+            buffer,
+            len,
+            offset: 0,
+        }
+    }
+
+    /// Create a new `BoolBuffer` of length `len` where all bits are unset (false).
+    pub fn new_unset(len: usize) -> Self {
+        let words = len.div_ceil(8);
+        let buffer = Buffer::zeroed(words);
+
+        Self {
+            buffer,
+            len,
+            offset: 0,
+        }
+    }
+
+    /// Invokes `f` with indexes `0..len` collecting the boolean results into a new `BitBuffer`
+    pub fn collect_bool<F: FnMut(usize) -> bool>(len: usize, mut f: F) -> Self {
+        let mut buffer = BufferMut::with_capacity(len.div_ceil(64) * 8);
+
+        let chunks = len / 64;
+        let remainder = len % 64;
+        for chunk in 0..chunks {
+            let mut packed = 0;
+            for bit_idx in 0..64 {
+                let i = bit_idx + chunk * 64;
+                packed |= (f(i) as u64) << bit_idx;
+            }
+
+            // SAFETY: Already allocated sufficient capacity
+            unsafe { buffer.push_unchecked(packed) }
+        }
+
+        if remainder != 0 {
+            let mut packed = 0;
+            for bit_idx in 0..remainder {
+                let i = bit_idx + chunks * 64;
+                packed |= (f(i) as u64) << bit_idx;
+            }
+
+            // SAFETY: Already allocated sufficient capacity
+            unsafe { buffer.push_unchecked(packed) }
+        }
+
+        buffer.truncate(len.div_ceil(8));
+
+        Self::new(
+            buffer
+                .freeze()
+                .into_byte_buffer()
+                .aligned(Alignment::of::<u8>()),
+            len,
+        )
+    }
+
+    /// Get the logical length of this `BoolBuffer`.
+    ///
+    /// This may differ from the physical length of the backing buffer, for example if it was
+    /// created using the `new_with_offset` constructor, or if it was sliced.
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    /// Returns `true` if the `BoolBuffer` is empty.
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Offset of the start of the buffer in bits.
+    #[inline]
+    pub fn offset(&self) -> usize {
+        self.offset
+    }
+
+    /// Get a reference to the underlying buffer.
+    #[inline]
+    pub fn inner(&self) -> &ByteBuffer {
+        &self.buffer
+    }
+
+    /// Retrieve the value at the given index.
+    ///
+    /// Panics if the index is out of bounds.
+    pub fn value(&self, index: usize) -> bool {
+        get_bit(&self.buffer, index + self.offset)
+    }
+
+    /// Retrieve the value at the given index without bounds checking
+    ///
+    /// # SAFETY
+    /// Caller must ensure that index is within the range of the buffer
+    pub unsafe fn value_unchecked(&self, index: usize) -> bool {
+        unsafe { get_bit_unchecked(&self.buffer, index + self.offset) }
+    }
+
+    /// Create a new zero-copy slice of this BoolBuffer that begins at the `start` index and extends
+    /// for `len` bits.
+    ///
+    /// Panics if the slice would extend beyond the end of the buffer.
+    pub fn slice(&self, range: Range<usize>) -> Self {
+        assert!(
+            range.len() <= self.len,
+            "slice from {} to {} exceeds len {}",
+            range.start,
+            range.end,
+            range.len()
+        );
+
+        Self::new_with_offset(self.buffer.clone(), range.len(), self.offset + range.start)
+    }
+
+    /// Slice any full bytes from the buffer, leaving the offset < 8.
+    pub fn shrink_offset(self) -> Self {
+        let bit_offset = self.offset % 8;
+        let len = self.len;
+        let buffer = self.into_inner();
+        BitBuffer::new_with_offset(buffer, len, bit_offset)
+    }
+
+    /// Access chunks of the buffer aligned to 8 byte boundary as [prefix, \<full chunks\>, suffix]
+    pub fn unaligned_chunks(&self) -> UnalignedBitChunks {
+        UnalignedBitChunks::new(self.buffer.clone(), self.offset, self.len)
+    }
+
+    /// Access chunks of the underlying buffer as 8 byte chunks with a final trailer
+    ///
+    /// If you're performing operations on a single buffer, prefer [BitBuffer::unaligned_chunks]
+    pub fn chunks(&self) -> BitChunks {
+        BitChunks::new(self.buffer.clone(), self.offset, self.len)
+    }
+
+    /// Get the number of set bits in the buffer.
+    pub fn true_count(&self) -> usize {
+        self.unaligned_chunks().count_ones()
+    }
+
+    /// Get the number of unset bits in the buffer.
+    pub fn false_count(&self) -> usize {
+        self.len - self.true_count()
+    }
+
+    /// Iterator over bits in the buffer
+    pub fn iter(&self) -> BitIterator {
+        BitIterator::new(self.buffer.clone(), self.offset, self.len)
+    }
+
+    /// Iterator over set indices of the underlying buffer
+    pub fn set_indices(&self) -> BitIndexIterator {
+        BitIndexIterator::new(self.buffer.clone(), self.offset, self.len)
+    }
+
+    /// Iterator over set slices of the underlying buffer
+    pub fn set_slices(&self) -> BitSliceIterator {
+        BitSliceIterator::new(self.buffer.clone(), self.offset, self.len)
+    }
+
+    /// Created a new BitBuffer with offset reset to 0
+    pub fn sliced(&self) -> Self {
+        if self.offset % 8 == 0 {
+            return Self::new(
+                self.buffer.slice(self.offset / 8..self.len.div_ceil(8)),
+                self.len,
+            );
+        }
+
+        Self::new(
+            bitwise_unary_op(self.buffer.clone(), self.offset, self.len, |a| a),
+            self.len,
+        )
+    }
+}
+
+// Conversions
+
+impl BitBuffer {
+    /// Consumes this `BoolBuffer` and returns the backing `Buffer<u8>` with any offset
+    /// and length information applied.
+    pub fn into_inner(self) -> ByteBuffer {
+        let word_start = self.offset / 8;
+        let word_end = (self.offset + self.len).div_ceil(8);
+
+        self.buffer.slice(word_start..word_end)
+    }
+
+    /// Get a mutable version of this `BitBuffer` along with bit offset in the first byte.
+    ///
+    /// If the caller doesn't hold only reference to the underlying buffer, a copy is created.
+    /// The second value of the tuple is a bit_offset of the first value in the first byte
+    pub fn into_mut(self) -> (BitBufferMut, usize) {
+        let bit_offset = self.offset % 8;
+        let len = self.len;
+        // TODO(robert): if we are copying here we can strip offset bits
+        let shrunk = self.into_inner().into_mut();
+        (BitBufferMut::from_buffer(shrunk, len), bit_offset)
+    }
+}
+
+impl From<&[bool]> for BitBuffer {
+    fn from(value: &[bool]) -> Self {
+        let mut buf = BitBufferMut::new_unset(value.len());
+        for (i, &v) in value.iter().enumerate() {
+            if v {
+                // SAFETY: i is in bounds
+                unsafe { buf.set_unchecked(i) }
+            }
+        }
+        buf.freeze()
+    }
+}
+
+impl From<Vec<bool>> for BitBuffer {
+    fn from(value: Vec<bool>) -> Self {
+        value.as_slice().into()
+    }
+}
+
+impl FromIterator<bool> for BitBuffer {
+    fn from_iter<T: IntoIterator<Item = bool>>(iter: T) -> Self {
+        let iter = iter.into_iter();
+        let (low, high) = iter.size_hint();
+        if let Some(len) = high {
+            let mut buf = BitBufferMut::new_unset(len);
+            for (i, v) in iter.enumerate() {
+                if v {
+                    // SAFETY: i is in bounds
+                    unsafe { buf.set_unchecked(i) }
+                }
+            }
+            buf.freeze()
+        } else {
+            let mut buf = BitBufferMut::new(low);
+            for v in iter {
+                buf.append(v);
+            }
+            buf.freeze()
+        }
+    }
+}
+
+impl BitOr for &BitBuffer {
+    type Output = BitBuffer;
+
+    fn bitor(self, rhs: Self) -> Self::Output {
+        self.clone() | rhs.clone()
+    }
+}
+
+impl BitOr for BitBuffer {
+    type Output = BitBuffer;
+
+    fn bitor(self, rhs: Self) -> Self::Output {
+        assert_eq!(self.len, rhs.len);
+        BitBuffer::new_with_offset(
+            bitwise_or(self.buffer, self.offset, rhs.buffer, rhs.offset, self.len),
+            self.len,
+            0,
+        )
+    }
+}
+
+impl BitAnd for &BitBuffer {
+    type Output = BitBuffer;
+
+    fn bitand(self, rhs: Self) -> Self::Output {
+        self.clone() & rhs.clone()
+    }
+}
+
+impl BitAnd for BitBuffer {
+    type Output = BitBuffer;
+
+    fn bitand(self, rhs: Self) -> Self::Output {
+        assert_eq!(self.len, rhs.len);
+        BitBuffer::new_with_offset(
+            bitwise_and(self.buffer, self.offset, rhs.buffer, rhs.offset, self.len),
+            self.len,
+            0,
+        )
+    }
+}
+
+impl Not for &BitBuffer {
+    type Output = BitBuffer;
+
+    fn not(self) -> Self::Output {
+        !self.clone()
+    }
+}
+
+impl Not for BitBuffer {
+    type Output = BitBuffer;
+
+    fn not(self) -> Self::Output {
+        BitBuffer::new_with_offset(bitwise_not(self.buffer, self.offset, self.len), self.len, 0)
+    }
+}
+
+impl BitXor for &BitBuffer {
+    type Output = BitBuffer;
+
+    fn bitxor(self, rhs: Self) -> Self::Output {
+        self.clone() ^ rhs.clone()
+    }
+}
+
+impl BitXor for BitBuffer {
+    type Output = BitBuffer;
+
+    fn bitxor(self, rhs: Self) -> Self::Output {
+        assert_eq!(self.len, rhs.len);
+        BitBuffer::new_with_offset(
+            bitwise_xor(self.buffer, self.offset, rhs.buffer, rhs.offset, self.len),
+            self.len,
+            0,
+        )
+    }
+}
+
+impl IntoIterator for &BitBuffer {
+    type Item = bool;
+    type IntoIter = BitIterator;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.iter()
+    }
+}
+
+impl IntoIterator for BitBuffer {
+    type Item = bool;
+    type IntoIter = BitIterator;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.iter()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::bit::BitBuffer;
+    use crate::{ByteBuffer, buffer};
+
+    #[test]
+    fn test_bool() {
+        // Create a new Buffer<u64> of length 1024 where the 8th bit is set.
+        let buffer: ByteBuffer = buffer![1 << 7; 1024];
+        let bools = BitBuffer::new(buffer, 1024 * 8);
+
+        // sanity checks
+        assert_eq!(bools.len(), 1024 * 8);
+        assert!(!bools.is_empty());
+        assert_eq!(bools.true_count(), 1024);
+        assert_eq!(bools.false_count(), 1024 * 7);
+
+        // Check all the values
+        for word in 0..1024 {
+            for bit in 0..8 {
+                if bit == 7 {
+                    assert!(bools.value(word * 8 + bit));
+                } else {
+                    assert!(!bools.value(word * 8 + bit));
+                }
+            }
+        }
+
+        // Slice the buffer to create a new subset view.
+        let sliced = bools.slice(64..72);
+
+        // sanity checks
+        assert_eq!(sliced.len(), 8);
+        assert!(!sliced.is_empty());
+        assert_eq!(sliced.true_count(), 1);
+        assert_eq!(sliced.false_count(), 7);
+
+        // Check all of the values like before
+        for bit in 0..8 {
+            if bit == 7 {
+                assert!(sliced.value(bit));
+            } else {
+                assert!(!sliced.value(bit));
+            }
+        }
+    }
+}
diff --git a/vortex-buffer/src/bit/buf_mut.rs b/vortex-buffer/src/bit/buf_mut.rs
new file mode 100644
index 00000000000..959adffb6c1
--- /dev/null
+++ b/vortex-buffer/src/bit/buf_mut.rs
@@ -0,0 +1,351 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::ops::Range;
+
+use bitvec::prelude::Lsb0;
+use bitvec::view::BitView;
+use vortex_error::VortexExpect;
+
+use crate::{BitBuffer, BufferMut, ByteBuffer, ByteBufferMut, buffer_mut, get_bit};
+
+/// A mutable bitset buffer that allows random access to individual bits for set and get.
+///
+///
+/// # Example
+/// ```
+/// use vortex_buffer::BitBufferMut;
+///
+/// let mut bools = BitBufferMut::new_unset(10);
+/// bools.set_to(9, true);
+/// for i in 0..9 {
+///    assert!(!bools.value(i));
+/// }
+/// assert!(bools.value(9));
+///
+/// // Freeze into a new bools vector.
+/// let bools = bools.freeze();
+/// ```
+///
+/// See also: [`crate::BitBuffer`].
+pub struct BitBufferMut {
+    buffer: ByteBufferMut,
+    len: usize,
+}
+
+impl BitBufferMut {
+    /// Create new bit buffer from given byte buffer and logical bit length
+    pub fn from_buffer(buffer: ByteBufferMut, len: usize) -> Self {
+        assert!(
+            len <= buffer.len() * 8,
+            "Buffer len {} is too short for the given length {len}",
+            buffer.len()
+        );
+        Self { buffer, len }
+    }
+
+    /// Create a new empty mutable bit buffer with requested capacity (in bits).
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            buffer: BufferMut::with_capacity(capacity.div_ceil(8)),
+            len: 0,
+        }
+    }
+
+    /// Create a new mutable buffer with requested `len` and all bits set to `true`.
+    pub fn new_set(len: usize) -> Self {
+        Self {
+            buffer: buffer_mut![0xFF; len.div_ceil(8)],
+            len,
+        }
+    }
+
+    /// Create a new mutable buffer with requested `len` and all bits set to `false`.
+    pub fn new_unset(len: usize) -> Self {
+        Self {
+            buffer: BufferMut::zeroed(len.div_ceil(8)),
+            len,
+        }
+    }
+
+    /// Create a new empty `BitBufferMut`.
+    pub fn empty() -> Self {
+        Self::new(0)
+    }
+
+    /// Get the current populated length of the buffer.
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    /// True if the buffer has length 0.
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+
+    /// Get the value at the requested index.
+    pub fn value(&self, index: usize) -> bool {
+        get_bit(&self.buffer, index)
+    }
+
+    /// Get the bit capacity of the buffer.
+    pub fn capacity(&self) -> usize {
+        self.buffer.capacity() * 8
+    }
+
+    /// Reserve additional bit capacity for the buffer.
+    pub fn reserve(&mut self, additional: usize) {
+        let capacity = self.len + additional;
+        if capacity > self.capacity() {
+            // convert differential to bytes
+            let additional = capacity.div_ceil(8) - self.buffer.len();
+            self.buffer.reserve(additional);
+        }
+    }
+
+    /// Set the bit at `index` to the given boolean value.
+    ///
+    /// This operation is checked so if `index` exceeds the buffer length, this will panic.
+    pub fn set_to(&mut self, index: usize, value: bool) {
+        if value {
+            self.set(index);
+        } else {
+            self.unset(index);
+        }
+    }
+
+    /// Set a position to `true`.
+    ///
+    /// This operation is checked so if `index` exceeds the buffer length, this will panic.
+    pub fn set(&mut self, index: usize) {
+        assert!(index < self.len, "index {index} exceeds len {}", self.len);
+
+        // SAFETY: checked by assertion
+        unsafe { self.set_unchecked(index) };
+    }
+
+    /// Set a position to `false`.
+    ///
+    /// This operation is checked so if `index` exceeds the buffer length, this will panic.
+    pub fn unset(&mut self, index: usize) {
+        assert!(index < self.len, "index {index} exceeds len {}", self.len);
+
+        // SAFETY: checked by assertion
+        unsafe { self.unset_unchecked(index) };
+    }
+
+    /// Set the bit at `index` to `true` without checking bounds.
+    ///
+    /// # Safety
+    ///
+    /// The caller must ensure that `index` does not exceed the largest bit index in the backing buffer.
+    pub unsafe fn set_unchecked(&mut self, index: usize) {
+        let word_index = index / 8;
+        let bit_index = index % 8;
+        // SAFETY: checked by caller
+        unsafe {
+            let word = self.buffer.as_mut_ptr().add(word_index);
+            word.write(*word | 1 << bit_index);
+        }
+    }
+
+    /// Unset the bit at `index` without checking bounds.
+    ///
+    /// # Safety
+    ///
+    /// The caller must ensure that `index` does not exceed the largest bit index in the backing buffer.
+    pub unsafe fn unset_unchecked(&mut self, index: usize) {
+        let word_index = index / 8;
+        let bit_index = index % 8;
+
+        // SAFETY: checked by caller
+        unsafe {
+            let word = self.buffer.as_mut_ptr().add(word_index);
+            word.write(*word & !(1 << bit_index));
+        }
+    }
+
+    /// Truncate the buffer to the given length.
+    pub fn truncate(&mut self, len: usize) {
+        if len > self.len {
+            return;
+        }
+
+        let new_len_bytes = len.div_ceil(8);
+        self.buffer.truncate(new_len_bytes);
+        self.len = len;
+
+        let remainder = self.len % 8;
+        if remainder != 0 {
+            let mask = (1u8 << remainder).wrapping_sub(1);
+            *self.buffer.as_mut().last_mut().vortex_expect("non empty") &= mask;
+        }
+    }
+
+    /// Append a new boolean into the bit buffer, incrementing the length.
+    ///
+    /// Panics if the buffer is full.
+    pub fn append(&mut self, value: bool) {
+        if value {
+            self.append_true()
+        } else {
+            self.append_false()
+        }
+    }
+
+    /// Append a new true value to the buffer.
+    ///
+    /// Panics if there is no remaining capacity.
+    pub fn append_true(&mut self) {
+        if self.len % 8 == 0 {
+            // Push a new word that starts with 1
+            self.buffer.push(1u8);
+        } else {
+            // Push a 1 bit into the current word.
+            let word = self.buffer.last_mut().vortex_expect("buffer is not empty");
+            *word |= 1 << (self.len % 8);
+        }
+
+        self.len += 1;
+    }
+
+    /// Append a new false value to the buffer.
+    ///
+    /// Panics if there is no remaining capacity.
+    pub fn append_false(&mut self) {
+        if self.len % 8 == 0 {
+            // push new word that starts with 0
+            self.buffer.push(0u8);
+        }
+
+        self.len += 1;
+    }
+    /// Append several boolean values into the bit buffer. After this operation,
+    /// the length will be incremented by `n`.
+    ///
+    /// Panics if the buffer does not have `n` slots left.
+    pub fn append_n(&mut self, value: bool, n: usize) {
+        match value {
+            true => {
+                let new_len = self.len + n;
+                let new_len_bytes = new_len.div_ceil(8);
+                let cur_remainder = self.len % 8;
+                let new_remainder = new_len % 8;
+
+                if cur_remainder != 0 {
+                    // Pad cur_remainder high bits with 1s
+                    *self
+                        .buffer
+                        .as_mut_slice()
+                        .last_mut()
+                        .vortex_expect("buffer is not empty") |= !((1 << cur_remainder) - 1);
+                }
+
+                // Push several full bytes.
+                if new_len_bytes > self.buffer.len() {
+                    // Push full bytes, except for the final byte.
+                    self.buffer.push_n(0xFF, new_len_bytes - self.buffer.len());
+                }
+
+                // Patch zeros into remainder of last byte pushed
+                if new_remainder > 0 {
+                    // Set the new_remainder LSB to 1
+                    *self
+                        .buffer
+                        .as_mut_slice()
+                        .last_mut()
+                        .vortex_expect("buffer is not empty") &= (1 << new_remainder) - 1;
+                }
+            }
+            false => {
+                let new_len = self.len + n;
+                let new_len_bytes = new_len.div_ceil(8);
+
+                // push new 0 bytes.
+                if new_len_bytes > self.buffer.len() {
+                    self.buffer.push_n(0, new_len_bytes - self.buffer.len());
+                }
+            }
+        }
+
+        self.len += n;
+    }
+
+    /// Append bits defined by range from values to this buffer
+    pub fn append_packed_range(&mut self, range: Range<usize>, values: &ByteBuffer) {
+        let bit_len = range.end - range.start;
+        self.buffer.reserve(bit_len.div_ceil(8));
+        // SAFETY: The copy below will populate the values
+        unsafe { self.buffer.set_len((self.len + bit_len).div_ceil(8)) };
+
+        let self_slice = self.buffer.as_mut_slice().view_bits_mut::<Lsb0>();
+        let other_slice = values.as_slice().view_bits::<Lsb0>();
+
+        let other_sliced = &other_slice[range.start..range.end];
+        self_slice[self.len..][..bit_len].copy_from_bitslice(other_sliced);
+        self.len += bit_len;
+    }
+
+    /// Append a [`BitBuffer`] to this [`BitBufferMut`]
+    pub fn append_buffer(&mut self, buffer: &BitBuffer) {
+        let buffer_range = buffer.offset()..buffer.offset() + buffer.len();
+        self.append_packed_range(buffer_range, buffer.inner())
+    }
+
+    /// Freeze the buffer in its current state into an immutable `BoolBuffer`.
+    pub fn freeze(self) -> BitBuffer {
+        BitBuffer::new(self.buffer.freeze().into_byte_buffer(), self.len)
+    }
+
+    /// Get the underlying bytes as a slice
+    pub fn as_slice(&self) -> &[u8] {
+        self.buffer.as_slice()
+    }
+
+    /// Get the underlying bytes as a mutable slice
+    pub fn as_mut_slice(&mut self) -> &mut [u8] {
+        self.buffer.as_mut_slice()
+    }
+}
+
+impl Default for BitBufferMut {
+    fn default() -> Self {
+        Self::new(0)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::bit::buf_mut::BitBufferMut;
+
+    #[test]
+    fn test_bits_mut() {
+        let mut bools = BitBufferMut::new_unset(10);
+        bools.set_to(0, true);
+        bools.set_to(9, true);
+
+        let bools = bools.freeze();
+        assert!(bools.value(0));
+        for i in 1..=8 {
+            assert!(!bools.value(i));
+        }
+        assert!(bools.value(9));
+    }
+
+    #[test]
+    fn test_append_n() {
+        let mut bools = BitBufferMut::new(10);
+        assert_eq!(bools.len(), 0);
+        assert!(bools.is_empty());
+
+        bools.append(true);
+        bools.append_n(false, 8);
+        bools.append_n(true, 1);
+
+        let bools = bools.freeze();
+
+        assert_eq!(bools.true_count(), 2);
+        assert!(bools.value(0));
+        assert!(bools.value(9));
+    }
+}
diff --git a/vortex-buffer/src/bit/mod.rs b/vortex-buffer/src/bit/mod.rs
new file mode 100644
index 00000000000..d427315b1f0
--- /dev/null
+++ b/vortex-buffer/src/bit/mod.rs
@@ -0,0 +1,35 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Packed bitmaps that can be used to store boolean values.
+//!
+//! This module provides a wrapper on top of the `Buffer` type to store mutable and immutable
+//! bitsets. The bitsets are stored in little-endian order, meaning that the least significant bit
+//! of the first byte is the first bit in the bitset.
+mod aligned;
+#[cfg(feature = "arrow")]
+mod arrow;
+mod buf;
+mod buf_mut;
+mod ops;
+mod unaligned;
+
+pub use aligned::BitChunks;
+pub use buf::*;
+pub use buf_mut::*;
+
+/// Get bit value at `index` out of `buf`
+#[inline]
+pub fn get_bit(buf: &[u8], index: usize) -> bool {
+    buf[index / 8] & (1 << (index % 8)) != 0
+}
+
+/// Get bit value at `index` out of `buf` without bounds checking
+///
+/// # Safety
+/// `index` must be between 0 and length of `buf`
+#[inline]
+pub unsafe fn get_bit_unchecked(buf: &[u8], index: usize) -> bool {
+    let byte = unsafe { buf.get_unchecked(index / 8) };
+    byte & (1 << (index % 8)) != 0
+}
diff --git a/vortex-buffer/src/bit/ops.rs b/vortex-buffer/src/bit/ops.rs
new file mode 100644
index 00000000000..d708857f604
--- /dev/null
+++ b/vortex-buffer/src/bit/ops.rs
@@ -0,0 +1,95 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use crate::{Alignment, BufferMut, ByteBuffer};
+
+pub fn bitwise_unary_op<F: FnMut(u64) -> u64>(
+    buffer: ByteBuffer,
+    offset: usize,
+    len: usize,
+    op: F,
+) -> ByteBuffer {
+    let mut result = BufferMut::<u64>::empty();
+    result.extend_trusted(buffer.bit_chunks(offset, len).iter().map(op));
+    result
+        .freeze()
+        .into_byte_buffer()
+        .aligned(Alignment::of::<u8>())
+}
+
+pub fn bitwise_binary_op<F: FnMut(u64, u64) -> u64>(
+    left_buffer: ByteBuffer,
+    left_offset: usize,
+    right_buffer: ByteBuffer,
+    right_offset: usize,
+    len: usize,
+    mut op: F,
+) -> ByteBuffer {
+    let mut result = BufferMut::<u64>::empty();
+    result.extend_trusted(
+        left_buffer
+            .bit_chunks(left_offset, len)
+            .iter()
+            .zip(right_buffer.bit_chunks(right_offset, len))
+            .map(|(l, r)| op(l, r)),
+    );
+    result
+        .freeze()
+        .into_byte_buffer()
+        .aligned(Alignment::of::<u8>())
+}
+
+pub fn bitwise_and(
+    left_buffer: ByteBuffer,
+    left_offset: usize,
+    right_buffer: ByteBuffer,
+    right_offset: usize,
+    len: usize,
+) -> ByteBuffer {
+    bitwise_binary_op(
+        left_buffer,
+        left_offset,
+        right_buffer,
+        right_offset,
+        len,
+        |l, r| l & r,
+    )
+}
+
+pub fn bitwise_or(
+    left_buffer: ByteBuffer,
+    left_offset: usize,
+    right_buffer: ByteBuffer,
+    right_offset: usize,
+    len: usize,
+) -> ByteBuffer {
+    bitwise_binary_op(
+        left_buffer,
+        left_offset,
+        right_buffer,
+        right_offset,
+        len,
+        |l, r| l | r,
+    )
+}
+
+pub fn bitwise_xor(
+    left_buffer: ByteBuffer,
+    left_offset: usize,
+    right_buffer: ByteBuffer,
+    right_offset: usize,
+    len: usize,
+) -> ByteBuffer {
+    bitwise_binary_op(
+        left_buffer,
+        left_offset,
+        right_buffer,
+        right_offset,
+        len,
+        |l, r| l ^ r,
+    )
+}
+
+pub fn bitwise_not(buffer: ByteBuffer, offset: usize, len: usize) -> ByteBuffer {
+    bitwise_unary_op(buffer, offset, len, |l| !l)
+}
diff --git a/vortex-buffer/src/bit/unaligned.rs b/vortex-buffer/src/bit/unaligned.rs
new file mode 100644
index 00000000000..3b604b33251
--- /dev/null
+++ b/vortex-buffer/src/bit/unaligned.rs
@@ -0,0 +1,351 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexExpect;
+
+use crate::trusted_len::TrustedLen;
+use crate::{Buffer, BufferIterator, ByteBuffer, get_bit_unchecked};
+
+#[inline]
+fn read_u64(input: &[u8]) -> u64 {
+    let len = input.len().min(8);
+    let mut buf = [0u8; 8];
+    buf[..len].copy_from_slice(input);
+    u64::from_le_bytes(buf)
+}
+
+#[inline]
+fn compute_prefix_mask(lead_padding: usize) -> u64 {
+    !((1 << lead_padding) - 1)
+}
+
+#[inline]
+fn compute_suffix_mask(len: usize, lead_padding: usize) -> (u64, usize) {
+    let trailing_bits = (len + lead_padding) % 64;
+
+    if trailing_bits == 0 {
+        return (u64::MAX, 0);
+    }
+
+    let trailing_padding = 64 - trailing_bits;
+    let suffix_mask = (1 << trailing_bits) - 1;
+    (suffix_mask, trailing_padding)
+}
+
+pub struct UnalignedBitChunks {
+    lead_padding: usize,
+    trailing_padding: usize,
+    prefix: Option<u64>,
+    chunks: Buffer<u64>,
+    suffix: Option<u64>,
+}
+
+impl UnalignedBitChunks {
+    pub fn new(buffer: ByteBuffer, offset: usize, len: usize) -> Self {
+        if len == 0 {
+            return Self {
+                lead_padding: 0,
+                trailing_padding: 0,
+                prefix: None,
+                chunks: Buffer::empty(),
+                suffix: None,
+            };
+        }
+        let byte_offset = offset / 8;
+        let offset_padding = offset % 8;
+        let bytes_len = (len + offset_padding).div_ceil(8);
+
+        let buffer = buffer.slice(byte_offset..byte_offset + bytes_len);
+
+        let prefix_mask = compute_prefix_mask(offset_padding);
+
+        // If less than 8 bytes, read into prefix
+        if buffer.len() <= 8 {
+            let (suffix_mask, trailing_padding) = compute_suffix_mask(len, offset_padding);
+            let prefix = read_u64(&buffer) & suffix_mask & prefix_mask;
+
+            return Self {
+                lead_padding: offset_padding,
+                trailing_padding,
+                prefix: Some(prefix),
+                chunks: Buffer::empty(),
+                suffix: None,
+            };
+        }
+
+        // If less than 16 bytes, read into prefix and suffix
+        if buffer.len() <= 16 {
+            let (suffix_mask, trailing_padding) = compute_suffix_mask(len, offset_padding);
+            let prefix = read_u64(&buffer[..8]) & prefix_mask;
+            let suffix = read_u64(&buffer[8..]) & suffix_mask;
+
+            return Self {
+                lead_padding: offset_padding,
+                trailing_padding,
+                prefix: Some(prefix),
+                chunks: Buffer::empty(),
+                suffix: Some(suffix),
+            };
+        }
+
+        let (prefix, mut chunks, suffix) = buffer.align_to::<u64>();
+        assert!(
+            prefix.len() < 8 && suffix.len() < 8,
+            "align_to did not return largest possible aligned slice"
+        );
+        let (alignment_padding, prefix) = match (offset_padding, prefix.is_empty()) {
+            (0, true) => (0, None),
+            (_, true) => {
+                let prefix = chunks[0] & prefix_mask;
+                chunks = chunks.slice(1..);
+                (0, Some(prefix))
+            }
+            (_, false) => {
+                let alignment_padding = (8 - prefix.len()) * 8;
+
+                let prefix = (read_u64(&prefix) & prefix_mask) << alignment_padding;
+                (alignment_padding, Some(prefix))
+            }
+        };
+
+        let lead_padding = offset_padding + alignment_padding;
+        let (suffix_mask, trailing_padding) = compute_suffix_mask(len, lead_padding);
+
+        let suffix = match (trailing_padding, suffix.is_empty()) {
+            (0, _) => None,
+            (_, true) => {
+                let suffix = chunks[chunks.len() - 1] & suffix_mask;
+                chunks = chunks.slice(..chunks.len() - 1);
+                Some(suffix)
+            }
+            (_, false) => Some(read_u64(&suffix) & suffix_mask),
+        };
+
+        Self {
+            lead_padding,
+            trailing_padding,
+            prefix,
+            chunks,
+            suffix,
+        }
+    }
+
+    pub fn iter(&self) -> UnalignedBitChunkIterator {
+        self.prefix
+            .into_iter()
+            .chain(self.chunks.clone())
+            .chain(self.suffix)
+    }
+
+    pub fn prefix_padding(&self) -> usize {
+        self.lead_padding
+    }
+
+    pub fn prefix(&self) -> Option<u64> {
+        self.prefix
+    }
+
+    pub fn suffix_padding(&self) -> usize {
+        self.trailing_padding
+    }
+
+    pub fn suffix(&self) -> Option<u64> {
+        self.suffix
+    }
+
+    pub fn count_ones(&self) -> usize {
+        self.iter().map(|x| x.count_ones() as usize).sum()
+    }
+}
+
+pub type UnalignedBitChunkIterator = core::iter::Chain<
+    core::iter::Chain<core::option::IntoIter<u64>, BufferIterator<u64>>,
+    core::option::IntoIter<u64>,
+>;
+
+/// Iterator over bits in the byte buffer
+pub struct BitIterator {
+    buffer: ByteBuffer,
+    current_offset: usize,
+    end_offset: usize,
+}
+
+impl BitIterator {
+    pub fn new(buffer: ByteBuffer, offset: usize, len: usize) -> Self {
+        let end_offset = offset + len;
+        assert!(
+            buffer.len() >= end_offset.div_ceil(8),
+            "Buffer {} too small for requested offset and len {}",
+            buffer.len(),
+            end_offset.div_ceil(8)
+        );
+
+        Self {
+            buffer,
+            current_offset: offset,
+            end_offset,
+        }
+    }
+}
+
+impl Iterator for BitIterator {
+    type Item = bool;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.current_offset == self.end_offset {
+            return None;
+        }
+        // SAFETY: current_offset is in bounds
+        let v = unsafe { get_bit_unchecked(&self.buffer, self.current_offset) };
+        self.current_offset += 1;
+        Some(v)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let remaining_bits = self.end_offset - self.current_offset;
+        (remaining_bits, Some(remaining_bits))
+    }
+}
+
+unsafe impl TrustedLen for BitIterator {}
+
+impl ExactSizeIterator for BitIterator {}
+
+impl DoubleEndedIterator for BitIterator {
+    fn next_back(&mut self) -> Option<Self::Item> {
+        if self.current_offset == self.end_offset {
+            return None;
+        }
+        self.end_offset -= 1;
+        // Safety: end_offset is in bounds
+        Some(unsafe { get_bit_unchecked(&self.buffer, self.end_offset) })
+    }
+}
+
+pub struct BitIndexIterator {
+    current_chunk: u64,
+    chunk_offset: i64,
+    iter: UnalignedBitChunkIterator,
+}
+
+impl BitIndexIterator {
+    pub fn new(buffer: ByteBuffer, offset: usize, len: usize) -> Self {
+        let chunks = UnalignedBitChunks::new(buffer, offset, len);
+        let mut iter = chunks.iter();
+        let current_chunk = iter.next().unwrap_or(0);
+        let chunk_offset = -(chunks.prefix_padding() as i64);
+
+        Self {
+            current_chunk,
+            chunk_offset,
+            iter,
+        }
+    }
+}
+
+impl Iterator for BitIndexIterator {
+    type Item = usize;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            if self.current_chunk != 0 {
+                let bit_pos = self.current_chunk.trailing_zeros();
+                self.current_chunk ^= 1 << bit_pos;
+                return Some(
+                    usize::try_from(self.chunk_offset + bit_pos as i64)
+                        .vortex_expect("bit index must be a usize"),
+                );
+            }
+
+            self.current_chunk = self.iter.next()?;
+            self.chunk_offset += 64;
+        }
+    }
+}
+
+pub struct BitSliceIterator {
+    iter: UnalignedBitChunkIterator,
+    len: usize,
+    current_offset: i64,
+    current_chunk: u64,
+}
+
+impl BitSliceIterator {
+    pub fn new(buffer: ByteBuffer, offset: usize, len: usize) -> Self {
+        let chunks = UnalignedBitChunks::new(buffer, offset, len);
+        let mut iter = chunks.iter();
+        let current_chunk = iter.next().unwrap_or(0);
+        let current_offset = -(chunks.prefix_padding() as i64);
+
+        Self {
+            iter,
+            len,
+            current_offset,
+            current_chunk,
+        }
+    }
+
+    /// Returns `Some((chunk_offset, bit_offset))` for the next chunk that has at
+    /// least one bit set, or None if there is no such chunk.
+    ///
+    /// Where `chunk_offset` is the bit offset to the current `u64` chunk
+    /// and `bit_offset` is the offset of the first `1` bit in that chunk
+    fn advance_to_set_bit(&mut self) -> Option<(i64, u32)> {
+        loop {
+            if self.current_chunk != 0 {
+                // Find the index of the first 1
+                let bit_pos = self.current_chunk.trailing_zeros();
+                return Some((self.current_offset, bit_pos));
+            }
+
+            self.current_chunk = self.iter.next()?;
+            self.current_offset += 64;
+        }
+    }
+}
+
+impl Iterator for BitSliceIterator {
+    type Item = (usize, usize);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.len == 0 {
+            return None;
+        }
+
+        let (start_chunk, start_bit) = self.advance_to_set_bit()?;
+
+        // Set bits up to start
+        self.current_chunk |= (1 << start_bit) - 1;
+
+        loop {
+            if self.current_chunk != u64::MAX {
+                // Find the index of the first 0
+                let end_bit = self.current_chunk.trailing_ones();
+
+                // Zero out up to end_bit
+                self.current_chunk &= !((1 << end_bit) - 1);
+
+                return Some((
+                    usize::try_from(start_chunk + start_bit as i64)
+                        .vortex_expect("bit offset must be a usize"),
+                    usize::try_from(self.current_offset + end_bit as i64)
+                        .vortex_expect("bit offset must be a usize"),
+                ));
+            }
+
+            match self.iter.next() {
+                Some(next) => {
+                    self.current_chunk = next;
+                    self.current_offset += 64;
+                }
+                None => {
+                    return Some((
+                        usize::try_from(start_chunk + start_bit as i64)
+                            .vortex_expect("bit offset must be a usize"),
+                        std::mem::take(&mut self.len),
+                    ));
+                }
+            }
+        }
+    }
+}

From 2683f53673cb1a2cd54ef7ee6b31d971a8f48bea Mon Sep 17 00:00:00 2001
From: Nicholas Gates <nick@nickgates.com>
Date: Mon, 13 Oct 2025 15:21:14 -0400
Subject: [PATCH 2/3] Port bit-buffer over

Signed-off-by: Nicholas Gates <nick@nickgates.com>
---
 vortex-buffer/src/bit/aligned.rs   |  9 ++++-
 vortex-buffer/src/bit/buf.rs       |  5 +--
 vortex-buffer/src/bit/buf_mut.rs   |  3 +-
 vortex-buffer/src/bit/mod.rs       |  4 +-
 vortex-buffer/src/bit/ops.rs       | 12 +++---
 vortex-buffer/src/bit/unaligned.rs |  3 +-
 vortex-buffer/src/buffer.rs        | 60 ++++++++++++++++++++++++++++++
 vortex-buffer/src/lib.rs           |  2 +
 vortex-buffer/src/trusted_len.rs   | 12 ++++--
 9 files changed, 93 insertions(+), 17 deletions(-)

diff --git a/vortex-buffer/src/bit/aligned.rs b/vortex-buffer/src/bit/aligned.rs
index 9e083181aed..a00b43f4dc2 100644
--- a/vortex-buffer/src/bit/aligned.rs
+++ b/vortex-buffer/src/bit/aligned.rs
@@ -16,6 +16,13 @@ pub struct BitChunks {
     remainder_len: usize,
 }
 
+impl ByteBuffer {
+    /// Returns an accessor which can be used to perform bitwise operations in u64 sized chunks
+    pub(super) fn bit_chunks(self, bit_offset: usize, bit_length: usize) -> BitChunks {
+        BitChunks::new(self.into_byte_buffer(), bit_offset, bit_length)
+    }
+}
+
 impl BitChunks {
     /// Construct new with given length and offset
     pub fn new(buffer: ByteBuffer, offset: usize, len: usize) -> Self {
@@ -60,7 +67,7 @@ impl BitChunks {
         result_bits & ((1 << self.remainder_len) - 1)
     }
 
-    /// Get an interator over the bitwise chunks including the trailer
+    /// Get an iterator over the bitwise chunks including the trailer
     pub fn iter(&self) -> PaddedBitChunksIterator {
         BitChunksIterator {
             buffer: self.buffer.clone(),
diff --git a/vortex-buffer/src/bit/buf.rs b/vortex-buffer/src/bit/buf.rs
index 21e1a081508..6575aa92188 100644
--- a/vortex-buffer/src/bit/buf.rs
+++ b/vortex-buffer/src/bit/buf.rs
@@ -6,9 +6,8 @@ use std::ops::{BitAnd, BitOr, BitXor, Not, Range};
 use crate::bit::aligned::BitChunks;
 use crate::bit::ops::{bitwise_and, bitwise_not, bitwise_or, bitwise_unary_op, bitwise_xor};
 use crate::bit::unaligned::{BitIndexIterator, BitIterator, BitSliceIterator, UnalignedBitChunks};
-use crate::{
-    Alignment, BitBufferMut, Buffer, BufferMut, ByteBuffer, buffer, get_bit, get_bit_unchecked,
-};
+use crate::bit::{get_bit, get_bit_unchecked};
+use crate::{Alignment, BitBufferMut, Buffer, BufferMut, ByteBuffer, buffer};
 
 /// An immutable bitset stored as a packed byte buffer.
 #[derive(Clone, Debug, Eq)]
diff --git a/vortex-buffer/src/bit/buf_mut.rs b/vortex-buffer/src/bit/buf_mut.rs
index 959adffb6c1..3b537257fe4 100644
--- a/vortex-buffer/src/bit/buf_mut.rs
+++ b/vortex-buffer/src/bit/buf_mut.rs
@@ -7,7 +7,8 @@ use bitvec::prelude::Lsb0;
 use bitvec::view::BitView;
 use vortex_error::VortexExpect;
 
-use crate::{BitBuffer, BufferMut, ByteBuffer, ByteBufferMut, buffer_mut, get_bit};
+use crate::bit::get_bit;
+use crate::{BitBuffer, BufferMut, ByteBuffer, ByteBufferMut, buffer_mut};
 
 /// A mutable bitset buffer that allows random access to individual bits for set and get.
 ///
diff --git a/vortex-buffer/src/bit/mod.rs b/vortex-buffer/src/bit/mod.rs
index d427315b1f0..79dc0bfb571 100644
--- a/vortex-buffer/src/bit/mod.rs
+++ b/vortex-buffer/src/bit/mod.rs
@@ -20,7 +20,7 @@ pub use buf_mut::*;
 
 /// Get bit value at `index` out of `buf`
 #[inline]
-pub fn get_bit(buf: &[u8], index: usize) -> bool {
+fn get_bit(buf: &[u8], index: usize) -> bool {
     buf[index / 8] & (1 << (index % 8)) != 0
 }
 
@@ -29,7 +29,7 @@ pub fn get_bit(buf: &[u8], index: usize) -> bool {
 /// # Safety
 /// `index` must be between 0 and length of `buf`
 #[inline]
-pub unsafe fn get_bit_unchecked(buf: &[u8], index: usize) -> bool {
+unsafe fn get_bit_unchecked(buf: &[u8], index: usize) -> bool {
     let byte = unsafe { buf.get_unchecked(index / 8) };
     byte & (1 << (index % 8)) != 0
 }
diff --git a/vortex-buffer/src/bit/ops.rs b/vortex-buffer/src/bit/ops.rs
index d708857f604..2f62f6cc0c3 100644
--- a/vortex-buffer/src/bit/ops.rs
+++ b/vortex-buffer/src/bit/ops.rs
@@ -3,7 +3,7 @@
 
 use crate::{Alignment, BufferMut, ByteBuffer};
 
-pub fn bitwise_unary_op<F: FnMut(u64) -> u64>(
+pub(super) fn bitwise_unary_op<F: FnMut(u64) -> u64>(
     buffer: ByteBuffer,
     offset: usize,
     len: usize,
@@ -17,7 +17,7 @@ pub fn bitwise_unary_op<F: FnMut(u64) -> u64>(
         .aligned(Alignment::of::<u8>())
 }
 
-pub fn bitwise_binary_op<F: FnMut(u64, u64) -> u64>(
+pub(super) fn bitwise_binary_op<F: FnMut(u64, u64) -> u64>(
     left_buffer: ByteBuffer,
     left_offset: usize,
     right_buffer: ByteBuffer,
@@ -39,7 +39,7 @@ pub fn bitwise_binary_op<F: FnMut(u64, u64) -> u64>(
         .aligned(Alignment::of::<u8>())
 }
 
-pub fn bitwise_and(
+pub(super) fn bitwise_and(
     left_buffer: ByteBuffer,
     left_offset: usize,
     right_buffer: ByteBuffer,
@@ -56,7 +56,7 @@ pub fn bitwise_and(
     )
 }
 
-pub fn bitwise_or(
+pub(super) fn bitwise_or(
     left_buffer: ByteBuffer,
     left_offset: usize,
     right_buffer: ByteBuffer,
@@ -73,7 +73,7 @@ pub fn bitwise_or(
     )
 }
 
-pub fn bitwise_xor(
+pub(super) fn bitwise_xor(
     left_buffer: ByteBuffer,
     left_offset: usize,
     right_buffer: ByteBuffer,
@@ -90,6 +90,6 @@ pub fn bitwise_xor(
     )
 }
 
-pub fn bitwise_not(buffer: ByteBuffer, offset: usize, len: usize) -> ByteBuffer {
+pub(super) fn bitwise_not(buffer: ByteBuffer, offset: usize, len: usize) -> ByteBuffer {
     bitwise_unary_op(buffer, offset, len, |l| !l)
 }
diff --git a/vortex-buffer/src/bit/unaligned.rs b/vortex-buffer/src/bit/unaligned.rs
index 3b604b33251..2a3d3ad7d3c 100644
--- a/vortex-buffer/src/bit/unaligned.rs
+++ b/vortex-buffer/src/bit/unaligned.rs
@@ -3,8 +3,9 @@
 
 use vortex_error::VortexExpect;
 
+use crate::bit::get_bit_unchecked;
 use crate::trusted_len::TrustedLen;
-use crate::{Buffer, BufferIterator, ByteBuffer, get_bit_unchecked};
+use crate::{Buffer, BufferIterator, ByteBuffer};
 
 #[inline]
 fn read_u64(input: &[u8]) -> u64 {
diff --git a/vortex-buffer/src/buffer.rs b/vortex-buffer/src/buffer.rs
index 157705916a4..6c5f51cc08b 100644
--- a/vortex-buffer/src/buffer.rs
+++ b/vortex-buffer/src/buffer.rs
@@ -448,6 +448,66 @@ impl<T> Buffer<T> {
             vortex_panic!("Buffer is not aligned to requested alignment {}", alignment)
         }
     }
+
+    /// Align the buffer to alignment of U
+    pub fn align_to<U>(mut self) -> (Buffer<T>, Buffer<U>, Buffer<T>) {
+        let offset = self.as_ptr().align_offset(align_of::<U>());
+        if offset > self.len() {
+            (
+                self,
+                Buffer::empty_aligned(Alignment::of::<U>()),
+                Buffer::empty_aligned(Alignment::of::<T>()),
+            )
+        } else {
+            let left = self.bytes.split_to(offset);
+            self.length -= offset;
+            let (us_len, _) = self.align_to_offsets::<U>();
+            let trailer = self.bytes.split_off(us_len * size_of::<U>());
+            (
+                Buffer::from_bytes_aligned(left, Alignment::of::<T>()),
+                Buffer::from_bytes_aligned(self.bytes, Alignment::of::<U>()),
+                Buffer::from_bytes_aligned(trailer, Alignment::of::<T>()),
+            )
+        }
+    }
+
+    /// Adapted from standard library slice::align_to_offsets
+    /// Function to calculate lengths of the middle and trailing slice for `align_to`.
+    fn align_to_offsets<U>(&self) -> (usize, usize) {
+        // What we're going to do about `rest` is figure out what multiple of `U`s we can put in the
+        // lowest number of `T`s. And how many `T`s we need for each such "multiple".
+        //
+        // Consider for example T=u8 U=u16. Then we can put 1 U in 2 Ts. Simple. Now, consider
+        // for example a case where size_of::<T> = 16, size_of::<U> = 24. We can put 2 Us in
+        // place of every 3 Ts in the `rest` slice. A bit more complicated.
+        //
+        // Formula to calculate this is:
+        //
+        // Us = lcm(size_of::<T>, size_of::<U>) / size_of::<U>
+        // Ts = lcm(size_of::<T>, size_of::<U>) / size_of::<T>
+        //
+        // Expanded and simplified:
+        //
+        // Us = size_of::<T> / gcd(size_of::<T>, size_of::<U>)
+        // Ts = size_of::<U> / gcd(size_of::<T>, size_of::<U>)
+        //
+        // Luckily since all this is constant-evaluated... performance here matters not!
+        const fn gcd(a: usize, b: usize) -> usize {
+            if b == 0 { a } else { gcd(b, a % b) }
+        }
+
+        // Explicitly wrap the function call in a const block so it gets
+        // constant-evaluated even in debug mode.
+        let gcd: usize = const { gcd(size_of::<T>(), size_of::<U>()) };
+        let ts: usize = size_of::<U>() / gcd;
+        let us: usize = size_of::<T>() / gcd;
+
+        // Armed with this knowledge, we can find how many `U`s we can fit!
+        let us_len = self.len() / ts * us;
+        // And how many `T`s will be in the trailing slice!
+        let ts_len = self.len() % ts;
+        (us_len, ts_len)
+    }
 }
 
 /// An iterator over Buffer elements.
diff --git a/vortex-buffer/src/lib.rs b/vortex-buffer/src/lib.rs
index 3780b9c5703..2bde672db93 100644
--- a/vortex-buffer/src/lib.rs
+++ b/vortex-buffer/src/lib.rs
@@ -48,6 +48,7 @@
 //! `arrow_buffer::OffsetBuffer`.
 
 pub use alignment::*;
+pub use bit::*;
 pub use buffer::*;
 pub use buffer_mut::*;
 pub use bytes::*;
@@ -57,6 +58,7 @@ pub use string::*;
 mod alignment;
 #[cfg(feature = "arrow")]
 mod arrow;
+mod bit;
 mod buffer;
 mod buffer_mut;
 mod bytes;
diff --git a/vortex-buffer/src/trusted_len.rs b/vortex-buffer/src/trusted_len.rs
index ba24e364b06..bacb3c4b27f 100644
--- a/vortex-buffer/src/trusted_len.rs
+++ b/vortex-buffer/src/trusted_len.rs
@@ -1,8 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::iter::Enumerate;
-
 use itertools::ProcessResults;
 
 /// Trait for all types which have a known upper-bound.
@@ -77,4 +75,12 @@ unsafe impl<'a, I, T: 'a, E: 'a> TrustedLen for ProcessResults<'a, I, E> where
 }
 
 // Enumerate
-unsafe impl<I, T> TrustedLen for Enumerate<I> where I: TrustedLen<Item = T> {}
+unsafe impl<I, T> TrustedLen for std::iter::Enumerate<I> where I: TrustedLen<Item = T> {}
+
+// Zip
+unsafe impl<T, U> TrustedLen for std::iter::Zip<T, U>
+where
+    T: TrustedLen,
+    U: TrustedLen,
+{
+}

From 467ea8ad83f6bf9b97d557811920de61d56aaa8f Mon Sep 17 00:00:00 2001
From: Nicholas Gates <nick@nickgates.com>
Date: Mon, 13 Oct 2025 15:44:41 -0400
Subject: [PATCH 3/3] Add some benchmarks

Signed-off-by: Nicholas Gates <nick@nickgates.com>
---
 vortex-buffer/Cargo.toml                  |   4 +
 vortex-buffer/benches/vortex_bitbuffer.rs | 253 ++++++++++++++++++++++
 vortex-buffer/src/bit/buf.rs              |  11 +-
 vortex-buffer/src/bit/buf_mut.rs          |   9 +-
 vortex-buffer/src/bit/mod.rs              |   9 +-
 vortex-buffer/src/bit/unaligned.rs        |   4 +-
 vortex-buffer/src/buffer.rs               |   3 +-
 7 files changed, 276 insertions(+), 17 deletions(-)
 create mode 100644 vortex-buffer/benches/vortex_bitbuffer.rs

diff --git a/vortex-buffer/Cargo.toml b/vortex-buffer/Cargo.toml
index c81972d4ba7..f91634d289f 100644
--- a/vortex-buffer/Cargo.toml
+++ b/vortex-buffer/Cargo.toml
@@ -46,3 +46,7 @@ divan = { workspace = true }
 [[bench]]
 name = "vortex_buffer"
 harness = false
+
+[[bench]]
+name = "vortex_bitbuffer"
+harness = false
diff --git a/vortex-buffer/benches/vortex_bitbuffer.rs b/vortex-buffer/benches/vortex_bitbuffer.rs
new file mode 100644
index 00000000000..91240885bb0
--- /dev/null
+++ b/vortex-buffer/benches/vortex_bitbuffer.rs
@@ -0,0 +1,253 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#![allow(clippy::unwrap_used)]
+
+use std::iter::Iterator;
+
+use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder};
+use divan::Bencher;
+use vortex_buffer::{BitBuffer, BitBufferMut};
+
+fn main() {
+    divan::main();
+}
+
+/// Wraps an arrow buffer so Divan can provide a nice name
+pub struct Arrow<T>(T);
+
+impl FromIterator<bool> for Arrow<BooleanBuffer> {
+    fn from_iter<I: IntoIterator<Item = bool>>(iter: I) -> Self {
+        Self(BooleanBuffer::from_iter(iter))
+    }
+}
+
+const INPUT_SIZE: &[usize] = &[128, 1024, 2048, 16_384, 65_536];
+
+#[divan::bench(
+    types = [Arrow<BooleanBuffer>, BitBuffer],
+    args = INPUT_SIZE,
+)]
+fn from_iter<B: FromIterator<bool>>(n: usize) {
+    B::from_iter((0..n).map(|i| i % 2 == 0));
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn append_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBufferMut::with_capacity(length))
+        .bench_refs(|buffer| {
+            for idx in 0..length {
+                buffer.append(divan::black_box(idx % 2 == 0));
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn append_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBufferBuilder::new(length)))
+        .bench_refs(|buffer| {
+            for idx in 0..length {
+                buffer.0.append(divan::black_box(idx % 2 == 0));
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn append_n_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBufferMut::with_capacity(length))
+        .bench_refs(|buffer| {
+            for _ in 0..100 {
+                buffer.append_n(divan::black_box(true), divan::black_box(length / 100));
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn append_n_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBufferBuilder::new(length)))
+        .bench_refs(|buffer| {
+            for _ in 0..100 {
+                buffer
+                    .0
+                    .append_n(divan::black_box(length / 100), divan::black_box(true));
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn value_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBuffer::from_iter((0..length).map(|i| i % 2 == 0)))
+        .bench_refs(|buffer| {
+            for idx in 0..length {
+                divan::black_box(buffer.value(idx));
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn value_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0))))
+        .bench_refs(|buffer| {
+            for idx in 0..length {
+                divan::black_box(buffer.0.value(idx));
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn slice_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBuffer::from_iter((0..length).map(|i| i % 2 == 0)))
+        .bench_values(|buffer| {
+            let mid = length / 2;
+            divan::black_box(buffer.slice(mid / 2..mid + mid / 2));
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn slice_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0))))
+        .bench_values(|buffer| {
+            let mid = length / 2;
+            divan::black_box(buffer.0.slice(mid / 2, mid / 2));
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn true_count_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBuffer::from_iter((0..length).map(|i| i % 2 == 0)))
+        .bench_refs(|buffer| {
+            divan::black_box(buffer.true_count());
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn true_count_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0))))
+        .bench_refs(|buffer| {
+            divan::black_box(buffer.0.count_set_bits());
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn bitwise_and_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| {
+            let a = BitBuffer::from_iter((0..length).map(|i| i % 2 == 0));
+            let b = BitBuffer::from_iter((0..length).map(|i| i % 3 == 0));
+            (a, b)
+        })
+        .bench_values(|(a, b)| {
+            divan::black_box(&a & &b);
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn bitwise_and_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| {
+            let a = Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0)));
+            let b = Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 3 == 0)));
+            (a, b)
+        })
+        .bench_values(|(a, b)| {
+            divan::black_box(&a.0 & &b.0);
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn bitwise_or_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| {
+            let a = BitBuffer::from_iter((0..length).map(|i| i % 2 == 0));
+            let b = BitBuffer::from_iter((0..length).map(|i| i % 3 == 0));
+            (a, b)
+        })
+        .bench_values(|(a, b)| {
+            divan::black_box(&a | &b);
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn bitwise_or_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| {
+            let a = Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0)));
+            let b = Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 3 == 0)));
+            (a, b)
+        })
+        .bench_values(|(a, b)| {
+            divan::black_box(&a.0 | &b.0);
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn bitwise_not_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBuffer::from_iter((0..length).map(|i| i % 2 == 0)))
+        .bench_values(|buffer| {
+            divan::black_box(!&buffer);
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn bitwise_not_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0))))
+        .bench_values(|buffer| {
+            divan::black_box(!&buffer.0);
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn iter_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBuffer::from_iter((0..length).map(|i| i % 2 == 0)))
+        .bench_refs(|buffer| {
+            for value in buffer.iter() {
+                divan::black_box(value);
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn iter_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0))))
+        .bench_refs(|buffer| {
+            for value in buffer.0.iter() {
+                divan::black_box(value);
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn set_indices_vortex_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| BitBuffer::from_iter((0..length).map(|i| i % 2 == 0)))
+        .bench_refs(|buffer| {
+            for idx in buffer.set_indices() {
+                divan::black_box(idx);
+            }
+        });
+}
+
+#[divan::bench(args = INPUT_SIZE)]
+fn set_indices_arrow_buffer(bencher: Bencher, length: usize) {
+    bencher
+        .with_inputs(|| Arrow(BooleanBuffer::from_iter((0..length).map(|i| i % 2 == 0))))
+        .bench_refs(|buffer| {
+            for idx in buffer.0.set_indices() {
+                divan::black_box(idx);
+            }
+        });
+}
diff --git a/vortex-buffer/src/bit/buf.rs b/vortex-buffer/src/bit/buf.rs
index 6575aa92188..2786bd2bb02 100644
--- a/vortex-buffer/src/bit/buf.rs
+++ b/vortex-buffer/src/bit/buf.rs
@@ -4,9 +4,9 @@
 use std::ops::{BitAnd, BitOr, BitXor, Not, Range};
 
 use crate::bit::aligned::BitChunks;
+use crate::bit::get_bit_unchecked;
 use crate::bit::ops::{bitwise_and, bitwise_not, bitwise_or, bitwise_unary_op, bitwise_xor};
 use crate::bit::unaligned::{BitIndexIterator, BitIterator, BitSliceIterator, UnalignedBitChunks};
-use crate::bit::{get_bit, get_bit_unchecked};
 use crate::{Alignment, BitBufferMut, Buffer, BufferMut, ByteBuffer, buffer};
 
 /// An immutable bitset stored as a packed byte buffer.
@@ -158,16 +158,19 @@ impl BitBuffer {
     /// Retrieve the value at the given index.
     ///
     /// Panics if the index is out of bounds.
+    #[inline]
     pub fn value(&self, index: usize) -> bool {
-        get_bit(&self.buffer, index + self.offset)
+        assert!(index < self.len);
+        unsafe { self.value_unchecked(index) }
     }
 
     /// Retrieve the value at the given index without bounds checking
     ///
     /// # SAFETY
     /// Caller must ensure that index is within the range of the buffer
+    #[inline]
     pub unsafe fn value_unchecked(&self, index: usize) -> bool {
-        unsafe { get_bit_unchecked(&self.buffer, index + self.offset) }
+        unsafe { get_bit_unchecked(self.buffer.as_ptr(), index + self.offset) }
     }
 
     /// Create a new zero-copy slice of this BoolBuffer that begins at the `start` index and extends
@@ -305,7 +308,7 @@ impl FromIterator<bool> for BitBuffer {
             }
             buf.freeze()
         } else {
-            let mut buf = BitBufferMut::new(low);
+            let mut buf = BitBufferMut::with_capacity(low);
             for v in iter {
                 buf.append(v);
             }
diff --git a/vortex-buffer/src/bit/buf_mut.rs b/vortex-buffer/src/bit/buf_mut.rs
index 3b537257fe4..ec6a60f10cc 100644
--- a/vortex-buffer/src/bit/buf_mut.rs
+++ b/vortex-buffer/src/bit/buf_mut.rs
@@ -46,7 +46,7 @@ impl BitBufferMut {
     }
 
     /// Create a new empty mutable bit buffer with requested capacity (in bits).
-    pub fn new(capacity: usize) -> Self {
+    pub fn with_capacity(capacity: usize) -> Self {
         Self {
             buffer: BufferMut::with_capacity(capacity.div_ceil(8)),
             len: 0,
@@ -71,7 +71,7 @@ impl BitBufferMut {
 
     /// Create a new empty `BitBufferMut`.
     pub fn empty() -> Self {
-        Self::new(0)
+        Self::with_capacity(0)
     }
 
     /// Get the current populated length of the buffer.
@@ -198,6 +198,7 @@ impl BitBufferMut {
     ///
     /// Panics if there is no remaining capacity.
     pub fn append_true(&mut self) {
+        // TODO(ngates): this is surely pretty slow.
         if self.len % 8 == 0 {
             // Push a new word that starts with 1
             self.buffer.push(1u8);
@@ -311,7 +312,7 @@ impl BitBufferMut {
 
 impl Default for BitBufferMut {
     fn default() -> Self {
-        Self::new(0)
+        Self::with_capacity(0)
     }
 }
 
@@ -335,7 +336,7 @@ mod tests {
 
     #[test]
     fn test_append_n() {
-        let mut bools = BitBufferMut::new(10);
+        let mut bools = BitBufferMut::with_capacity(10);
         assert_eq!(bools.len(), 0);
         assert!(bools.is_empty());
 
diff --git a/vortex-buffer/src/bit/mod.rs b/vortex-buffer/src/bit/mod.rs
index 79dc0bfb571..bc1c5a3fca4 100644
--- a/vortex-buffer/src/bit/mod.rs
+++ b/vortex-buffer/src/bit/mod.rs
@@ -19,7 +19,7 @@ pub use buf::*;
 pub use buf_mut::*;
 
 /// Get bit value at `index` out of `buf`
-#[inline]
+#[inline(always)]
 fn get_bit(buf: &[u8], index: usize) -> bool {
     buf[index / 8] & (1 << (index % 8)) != 0
 }
@@ -28,8 +28,7 @@ fn get_bit(buf: &[u8], index: usize) -> bool {
 ///
 /// # Safety
 /// `index` must be between 0 and length of `buf`
-#[inline]
-unsafe fn get_bit_unchecked(buf: &[u8], index: usize) -> bool {
-    let byte = unsafe { buf.get_unchecked(index / 8) };
-    byte & (1 << (index % 8)) != 0
+#[inline(always)]
+unsafe fn get_bit_unchecked(buf: *const u8, index: usize) -> bool {
+    (unsafe { *buf.add(index / 8) } & (1 << (index % 8))) != 0
 }
diff --git a/vortex-buffer/src/bit/unaligned.rs b/vortex-buffer/src/bit/unaligned.rs
index 2a3d3ad7d3c..3bfc9de583d 100644
--- a/vortex-buffer/src/bit/unaligned.rs
+++ b/vortex-buffer/src/bit/unaligned.rs
@@ -197,7 +197,7 @@ impl Iterator for BitIterator {
             return None;
         }
         // SAFETY: current_offset is in bounds
-        let v = unsafe { get_bit_unchecked(&self.buffer, self.current_offset) };
+        let v = unsafe { get_bit_unchecked(self.buffer.as_ptr(), self.current_offset) };
         self.current_offset += 1;
         Some(v)
     }
@@ -219,7 +219,7 @@ impl DoubleEndedIterator for BitIterator {
         }
         self.end_offset -= 1;
         // Safety: end_offset is in bounds
-        Some(unsafe { get_bit_unchecked(&self.buffer, self.end_offset) })
+        Some(unsafe { get_bit_unchecked(self.buffer.as_ptr(), self.end_offset) })
     }
 }
 
diff --git a/vortex-buffer/src/buffer.rs b/vortex-buffer/src/buffer.rs
index 6c5f51cc08b..4c5752d70d3 100644
--- a/vortex-buffer/src/buffer.rs
+++ b/vortex-buffer/src/buffer.rs
@@ -221,9 +221,8 @@ impl<T> Buffer<T> {
     /// Returns a slice over the buffer of elements of type T.
     #[inline(always)]
     pub fn as_slice(&self) -> &[T] {
-        let raw_slice = self.bytes.as_ref();
         // SAFETY: alignment of Buffer is checked on construction
-        unsafe { std::slice::from_raw_parts(raw_slice.as_ptr().cast(), self.length) }
+        unsafe { std::slice::from_raw_parts(self.bytes.as_ptr().cast(), self.length) }
     }
 
     /// Returns an iterator over the buffer of elements of type T.