Benchmarks and performance improvement for parquet boolean reader (#10196)

# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->

- Closes #10195.

# Rationale for this change

Adds benchmarks and a performance improvements for reading boolean
arrays from parquet.

The optimized path for building a `Buffer` from `Vec<bool>` already
existed, using it yields a ~25% performance improvement on the new
benchmarks.

<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->

# Are these changes tested?

<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?

If this PR claims a performance improvement, please include evidence
such as benchmark results.
-->

# Are there any user-facing changes?

<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.

If there are any breaking changes to public APIs, please call them out.
-->

Author: jhorstmann

parquet/benches/arrow_reader.rs

@@ -35,7 +35,7 @@ use parquet::{
     arrow::array_reader::ArrayReader,
     basic::Encoding,
     column::page::PageIterator,
-    data_type::{ByteArrayType, Int32Type, Int64Type},
+    data_type::{BoolType, ByteArrayType, Int32Type, Int64Type},
     schema::types::{ColumnDescPtr, SchemaDescPtr},
 };
 use rand::distr::uniform::SampleUniform;
@@ -105,6 +105,8 @@ fn build_test_schema() -> SchemaDescPtr {
                     optional FIXED_LEN_BYTE_ARRAY(32) element;
                 }
             }
+            REQUIRED BOOLEAN mandatory_bool_leaf;
+            OPTIONAL BOOLEAN optional_bool_leaf;
         }
         ";
     parse_message_type(message_type)
@@ -337,6 +339,47 @@ where
     InMemoryPageIterator::new(pages)
 }
 
+fn build_encoded_bool_page_iterator(
+    column_desc: ColumnDescPtr,
+    null_density: f32,
+    encoding: Encoding,
+) -> impl PageIterator + Clone {
+    let max_def_level = column_desc.max_def_level();
+    let max_rep_level = column_desc.max_rep_level();
+    let rep_levels = vec![0; VALUES_PER_PAGE];
+    let mut rng = seedable_rng();
+    let mut pages: Vec<Vec<parquet::column::page::Page>> = Vec::new();
+    for _i in 0..NUM_ROW_GROUPS {
+        let mut column_chunk_pages = Vec::new();
+        for _j in 0..PAGES_PER_GROUP {
+            // generate page
+            let mut values = Vec::with_capacity(VALUES_PER_PAGE);
+            let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
+            for _k in 0..VALUES_PER_PAGE {
+                let def_level = if rng.random::<f32>() < null_density {
+                    max_def_level - 1
+                } else {
+                    max_def_level
+                };
+                if def_level == max_def_level {
+                    let value = rng.random_bool(0.5);
+                    values.push(value);
+                }
+                def_levels.push(def_level);
+            }
+            let mut page_builder =
+                DataPageBuilderImpl::new(column_desc.clone(), values.len() as u32, true);
+            page_builder.add_rep_levels(max_rep_level, &rep_levels);
+            page_builder.add_def_levels(max_def_level, &def_levels);
+            page_builder.add_values::<BoolType>(encoding, &values);
+            column_chunk_pages.push(page_builder.consume());
+        }
+        pages.push(column_chunk_pages);
+    }
+
+    InMemoryPageIterator::new(pages)
+}
+
 fn build_delta_encoded_incr_primitive_page_iterator<T>(
     column_desc: ColumnDescPtr,
     null_density: f32,
@@ -883,6 +926,16 @@ fn create_primitive_array_reader(
             .unwrap();
             Box::new(reader)
         }
+        Type::BOOLEAN => {
+            let reader = PrimitiveArrayReader::<BoolType>::new(
+                Box::new(page_iterator),
+                column_desc,
+                None,
+                DEFAULT_BATCH_SIZE,
+            )
+            .unwrap();
+            Box::new(reader)
+        }
         _ => unreachable!(),
     }
 }
@@ -1575,6 +1628,47 @@ fn bench_primitive<T>(
     });
 }
 
+fn bench_boolean(
+    group: &mut BenchmarkGroup<WallTime>,
+    mandatory_column_desc: &ColumnDescPtr,
+    optional_column_desc: &ColumnDescPtr,
+) {
+    let mut count: usize = 0;
+
+    // plain encoded, no NULLs
+    let data =
+        build_encoded_bool_page_iterator(mandatory_column_desc.clone(), 0.0, Encoding::PLAIN);
+    group.bench_function("plain encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_primitive_array_reader(data.clone(), mandatory_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let data = build_encoded_bool_page_iterator(optional_column_desc.clone(), 0.0, Encoding::PLAIN);
+    group.bench_function("plain encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_primitive_array_reader(data.clone(), optional_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // plain encoded, half NULLs
+    let data = build_encoded_bool_page_iterator(optional_column_desc.clone(), 0.5, Encoding::PLAIN);
+    group.bench_function("plain encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_primitive_array_reader(data.clone(), optional_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+}
+
 // Benchmark reading a struct with a single primitive field.
 // No need to bench all encodings for the data, as that should already be covered by `bench_primitive`.
 // The only performance difference should be caused by the additional definition level.
@@ -1801,6 +1895,8 @@ fn add_benches(c: &mut Criterion) {
     let optional_struct_optional_in32_column_desc = schema.column(40);
     let int32_list_desc = schema.column(41);
     let fixed32_list_desc = schema.column(42);
+    let mandatory_bool_column_desc = schema.column(43);
+    let optional_bool_column_desc = schema.column(44);
 
     // primitive / int32 benchmarks
     // =============================
@@ -1894,6 +1990,15 @@ fn add_benches(c: &mut Criterion) {
     );
     group.finish();
 
+    // boolean benchmarks
+    let mut group = c.benchmark_group("arrow_array_reader/BooleanArray");
+    bench_boolean(
+        &mut group,
+        &mandatory_bool_column_desc,
+        &optional_bool_column_desc,
+    );
+    group.finish();
+
     let mut group = c.benchmark_group("arrow_array_reader/struct/Int32Array");
     bench_struct_primitive::<Int32Type>(
         &mut group,

parquet/src/arrow/array_reader/primitive_array.rs

@@ -56,7 +56,7 @@ native_buffer!(i8, i16, i32, i64, u8, u16, u32, u64, f32, f64);
 
 impl IntoBuffer for Vec<bool> {
     fn into_buffer(self, _target_type: &ArrowType) -> Buffer {
-        BooleanBuffer::from_iter(self).into_inner()
+        BooleanBuffer::from(self.as_slice()).into_inner()
     }
 }