apache
diff --git a/‎arrow-buffer/src/builder/boolean.rs‎
Lines changed: 136 additions & 0 deletions b/‎arrow-buffer/src/builder/boolean.rs‎
Lines changed: 136 additions & 0 deletions
diff --git a/‎parquet/benches/arrow_reader.rs‎
Lines changed: 30 additions & 4 deletions b/‎parquet/benches/arrow_reader.rs‎
Lines changed: 30 additions & 4 deletions
diff --git a/‎parquet/benches/arrow_reader_peak_memory.rs‎
Lines changed: 32 additions & 10 deletions b/‎parquet/benches/arrow_reader_peak_memory.rs‎
Lines changed: 32 additions & 10 deletions
@@ -186,6 +186,46 @@ impl BooleanBufferBuilder {
         }
     }
 
+    /// Appends the low `count` bits from `word` into the buffer.
+    ///
+    /// `word` is treated as a packed LSB-first bitmap. Only the lowest
+    /// `count` bits are appended; higher bits are ignored.
+    ///
+    /// This is significantly faster than calling [`Self::append`] in a
+    /// loop when the caller already has bits packed into a `u64`.
+    #[inline]
+    pub fn append_word(&mut self, word: u64, count: usize) {
+        debug_assert!(count <= 64);
+        let mask = (u64::MAX >> ((64 - count) & 63)) * ((count != 0) as u64);
+        let word = word & mask;
+
+        let new_len = self.len + count;
+        let new_len_bytes = bit_util::ceil(new_len, 8);
+        if new_len_bytes > self.buffer.len() {
+            self.buffer.resize(new_len_bytes, 0);
+        }
+
+        let bit_offset = self.len & 7;
+        let byte_start = self.len / 8;
+        let buf = self.buffer.as_slice_mut();
+
+        // Shift the word to align with the bit offset within the
+        // current byte, then OR it into the buffer. The shift merges
+        // correctly with any existing bits in the partial trailing byte.
+        let shifted = word << bit_offset;
+        let shifted_bytes = shifted.to_le_bytes();
+        let bytes_to_write = bit_util::ceil(count + bit_offset, 8).min(8);
+        for i in 0..bytes_to_write {
+            buf[byte_start + i] |= shifted_bytes[i];
+        }
+        // When bit_offset > 0 the shift can overflow into a 9th byte.
+        if bit_offset > 0 && count + bit_offset > 64 {
+            buf[byte_start + 8] |= (word >> (64 - bit_offset)) as u8;
+        }
+
+        self.len = new_len;
+    }
+
     /// Appends n `additional` bits of value `v` into the buffer
     #[inline]
     pub fn append_n(&mut self, additional: usize, v: bool) {
@@ -619,6 +659,102 @@ mod tests {
         }
     }
 
+    /// Helper: build via append_word and verify against bit-by-bit append.
+    fn check_append_word(initial_bits: usize, word: u64, count: usize) {
+        let mut got = BooleanBufferBuilder::new(0);
+        let mut expected = BooleanBufferBuilder::new(0);
+        got.append_n(initial_bits, true);
+        expected.append_n(initial_bits, true);
+        got.append_word(word, count);
+        for i in 0..count {
+            expected.append(word & (1 << i) != 0);
+        }
+        assert_eq!(got.len(), expected.len());
+        assert_eq!(got.finish(), expected.finish());
+    }
+
+    #[test]
+    fn test_append_word_zero_count() {
+        check_append_word(0, u64::MAX, 0);
+        check_append_word(3, u64::MAX, 0);
+    }
+
+    #[test]
+    fn test_append_word_aligned() {
+        for count in [1, 5, 8, 17, 64] {
+            check_append_word(0, 0xDEAD_BEEF_CAFE_BABE, count);
+        }
+    }
+
+    #[test]
+    fn test_append_word_unaligned() {
+        for offset in 1..=7 {
+            check_append_word(offset, 0xDEAD_BEEF_CAFE_BABE, 13);
+        }
+    }
+
+    #[test]
+    fn test_append_word_overflow_9th_byte() {
+        check_append_word(3, u64::MAX, 64);
+        check_append_word(7, 0xA5A5_A5A5_A5A5_A5A5, 64);
+    }
+
+    #[test]
+    fn test_append_word_small_counts() {
+        check_append_word(0, 0b1, 1);
+        check_append_word(0, 0b1010101, 7);
+        check_append_word(3, 0b1, 1);
+        check_append_word(3, 0b1010101, 7);
+    }
+
+    #[test]
+    fn test_append_word_ignores_high_bits_before_later_appends() {
+        let mut builder = BooleanBufferBuilder::new(0);
+        builder.append_word(0b10, 1);
+        builder.append(false);
+
+        let finished = builder.finish();
+        assert_eq!(finished.len(), 2);
+        assert!(!finished.value(0));
+        assert!(!finished.value(1));
+    }
+
+    #[test]
+    fn test_append_word_full_word() {
+        check_append_word(0, u64::MAX, 64);
+        check_append_word(0, 0, 64);
+    }
+
+    #[test]
+    fn test_append_word_sequential() {
+        let mut got = BooleanBufferBuilder::new(0);
+        let mut expected = BooleanBufferBuilder::new(0);
+        for (word, count) in [(0b1010u64, 4), (0b111u64, 3), (0u64, 5), (u64::MAX, 64)] {
+            got.append_word(word, count);
+            for i in 0..count {
+                expected.append(word & (1 << i) != 0);
+            }
+        }
+        assert_eq!(got.finish(), expected.finish());
+    }
+
+    #[test]
+    fn test_append_word_mixed_with_append() {
+        let mut got = BooleanBufferBuilder::new(0);
+        let mut expected = BooleanBufferBuilder::new(0);
+        got.append(true);
+        expected.append(true);
+        got.append_word(0b1100, 4);
+        for i in 0..4 {
+            expected.append(0b1100u64 & (1 << i) != 0);
+        }
+        got.append(false);
+        expected.append(false);
+        got.append_word(0xFF, 8);
+        expected.append_n(8, true);
+        assert_eq!(got.finish(), expected.finish());
+    }
+
     #[test]
     fn test_extend_misaligned_end() {
         for len in 1..130 {
 
@@ -785,12 +785,15 @@ fn create_int32_list_reader(
             column_desc,
             None,
             DEFAULT_BATCH_SIZE,
+            Some(2),
         )
         .unwrap(),
     ) as Box<dyn ArrayReader>;
     let field = Field::new_list_field(DataType::Int32, true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(items, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        items, data_type, 2, 1, true, None,
+    ))
 }
 
 const FIXED_BYTE_LEN: usize = 32;
@@ -858,11 +861,14 @@ fn create_fixed32_list_reader(
         column_desc,
         None,
         DEFAULT_BATCH_SIZE,
+        Some(2),
     )
     .unwrap();
     let field = Field::new_list_field(DataType::FixedSizeBinary(FIXED_BYTE_LEN as i32), true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(items, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        items, data_type, 2, 1, true, None,
+    ))
 }
 
 fn bench_array_reader(mut array_reader: Box<dyn ArrayReader>) -> usize {
@@ -912,6 +918,7 @@ fn create_primitive_array_reader(
                 column_desc,
                 None,
                 DEFAULT_BATCH_SIZE,
+                None,
             )
             .unwrap();
             Box::new(reader)
@@ -922,6 +929,7 @@ fn create_primitive_array_reader(
                 column_desc,
                 None,
                 DEFAULT_BATCH_SIZE,
+                None,
             )
             .unwrap();
             Box::new(reader)
@@ -932,6 +940,7 @@ fn create_primitive_array_reader(
                 column_desc,
                 None,
                 DEFAULT_BATCH_SIZE,
+                None,
             )
             .unwrap();
             Box::new(reader)
@@ -951,6 +960,7 @@ fn create_f16_by_bytes_reader(
             column_desc,
             None,
             DEFAULT_BATCH_SIZE,
+            None,
         )
         .unwrap(),
         _ => unimplemented!(),
@@ -968,13 +978,15 @@ fn create_decimal_by_bytes_reader(
             column_desc,
             None,
             DEFAULT_BATCH_SIZE,
+            None,
         )
         .unwrap(),
         Type::FIXED_LEN_BYTE_ARRAY => make_fixed_len_byte_array_reader(
             Box::new(page_iterator),
             column_desc,
             None,
             DEFAULT_BATCH_SIZE,
+            None,
         )
         .unwrap(),
         _ => unimplemented!(),
@@ -990,6 +1002,7 @@ fn create_fixed_len_byte_array_reader(
         column_desc,
         None,
         DEFAULT_BATCH_SIZE,
+        None,
     )
     .unwrap()
 }
@@ -1003,6 +1016,7 @@ fn create_byte_array_reader(
         column_desc,
         None,
         DEFAULT_BATCH_SIZE,
+        None,
     )
     .unwrap()
 }
@@ -1016,6 +1030,7 @@ fn create_byte_view_array_reader(
         column_desc,
         None,
         DEFAULT_BATCH_SIZE,
+        None,
     )
     .unwrap()
 }
@@ -1029,6 +1044,7 @@ fn create_string_view_byte_array_reader(
         column_desc,
         None,
         DEFAULT_BATCH_SIZE,
+        None,
     )
     .unwrap()
 }
@@ -1045,6 +1061,7 @@ fn create_string_byte_array_dictionary_reader(
         column_desc,
         Some(arrow_type),
         DEFAULT_BATCH_SIZE,
+        None,
     )
     .unwrap()
 }
@@ -1053,10 +1070,19 @@ fn create_string_list_reader(
     page_iterator: impl PageIterator + 'static,
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
-    let items = create_byte_array_reader(page_iterator, column_desc);
+    let items = make_byte_array_reader(
+        Box::new(page_iterator),
+        column_desc,
+        None,
+        DEFAULT_BATCH_SIZE,
+        Some(2),
+    )
+    .unwrap();
     let field = Field::new_list_field(DataType::Utf8, true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(items, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        items, data_type, 2, 1, true, None,
+    ))
 }
 
 fn bench_byte_decimal<T>(
 
@@ -443,46 +443,68 @@ fn int32_list_wrapper(
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
     let child: Box<dyn ArrayReader> = Box::new(
-        PrimitiveArrayReader::<Int32Type>::new(pages, column_desc, None, DEFAULT_BATCH_SIZE)
-            .unwrap(),
+        PrimitiveArrayReader::<Int32Type>::new(
+            pages,
+            column_desc,
+            None,
+            DEFAULT_BATCH_SIZE,
+            Some(2),
+        )
+        .unwrap(),
     );
     let field = Field::new_list_field(DataType::Int32, true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(child, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        child, data_type, 2, 1, true, None,
+    ))
 }
 
 fn double_list_wrapper(
     pages: Box<dyn PageIterator>,
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
     let child: Box<dyn ArrayReader> = Box::new(
-        PrimitiveArrayReader::<DoubleType>::new(pages, column_desc, None, DEFAULT_BATCH_SIZE)
-            .unwrap(),
+        PrimitiveArrayReader::<DoubleType>::new(
+            pages,
+            column_desc,
+            None,
+            DEFAULT_BATCH_SIZE,
+            Some(2),
+        )
+        .unwrap(),
     );
     let field = Field::new_list_field(DataType::Float64, true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(child, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        child, data_type, 2, 1, true, None,
+    ))
 }
 
 fn fixed32_list_wrapper(
     pages: Box<dyn PageIterator>,
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
     let child =
-        make_fixed_len_byte_array_reader(pages, column_desc, None, DEFAULT_BATCH_SIZE).unwrap();
+        make_fixed_len_byte_array_reader(pages, column_desc, None, DEFAULT_BATCH_SIZE, Some(2))
+            .unwrap();
     let field = Field::new_list_field(DataType::FixedSizeBinary(FIXED_BYTE_LEN as i32), true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(child, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        child, data_type, 2, 1, true, None,
+    ))
 }
 
 fn string_list_wrapper(
     pages: Box<dyn PageIterator>,
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
-    let child = make_byte_array_reader(pages, column_desc, None, DEFAULT_BATCH_SIZE).unwrap();
+    let child =
+        make_byte_array_reader(pages, column_desc, None, DEFAULT_BATCH_SIZE, Some(2)).unwrap();
     let field = Field::new_list_field(DataType::Utf8, true);
     let data_type = DataType::List(Arc::new(field));
-    Box::new(ListArrayReader::<i32>::new(child, data_type, 2, 1, true))
+    Box::new(ListArrayReader::<i32>::new(
+        child, data_type, 2, 1, true, None,
+    ))
 }
 
 fn add_benches<M: Measurement>(c: &mut Criterion<M>, measurement_name: &str) {