[Parquet] Reduce reallocations when reading StringView in parquet

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

I am profiling clickbench query 10 with predicate pushdown enabled as part of

• Enable parquet filter pushdown (filter_pushdown) by default datafusion#3463

samply record -- /Users/andrewlamb/Software/datafusion2/target/profiling/datafusion-cli   -f q.sql  > /dev/null  2>&1

SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;

While looking at the profile, I noticed that 7% of the time is spent in allocating / regrowing vectors (aka reallocating and copying)

Describe the solution you'd like
Avoid the time spent regrowing these vectors

It appears that the vectors in question are part of the ViewBuffer struct:

arrow-rs/parquet/src/arrow/buffer/view_buffer.rs

Lines 30 to 33 in 02fa779

	pub struct ViewBuffer {
	pub views: Vec<u128>,
	pub buffers: Vec<Buffer>,
	}

Describe alternatives you've considered
Since we know how many views will be in each output buffer, we could create the ViewBuffers with the correct size initially

Something like like

ViewBuffers::with_capacity

Additional context

[alamb]

Here is one potential change:

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index f881690f805..71e517ffb61 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -162,11 +162,12 @@ impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
             ));
         }

-        let mut buffer = ViewBuffer::default();
+        let num_values = num_values as usize;
+        let mut buffer = ViewBuffer::with_capacity(num_values);
         let mut decoder = ByteViewArrayDecoderPlain::new(
             buf,
-            num_values as usize,
-            Some(num_values as usize),
+            num_values,
+            Some(num_values),
             self.validate_utf8,
         );
         decoder.read(&mut buffer, usize::MAX)?;
diff --git a/parquet/src/arrow/buffer/view_buffer.rs b/parquet/src/arrow/buffer/view_buffer.rs
index 2802f97f8f3..fab5a2488b9 100644
--- a/parquet/src/arrow/buffer/view_buffer.rs
+++ b/parquet/src/arrow/buffer/view_buffer.rs
@@ -33,6 +33,14 @@ pub struct ViewBuffer {
 }

 impl ViewBuffer {
+    /// Create a new ViewBuffer with capacity for the specified number of views
+    pub fn with_capacity(capacity: usize) -> Self {
+        Self {
+            views: Vec::with_capacity(capacity),
+            buffers: Vec::new(),
+        }
+    }
+
     pub fn is_empty(&self) -> bool {
         self.views.is_empty()
     }
@@ -111,7 +119,6 @@ impl ValuesBuffer for ViewBuffer {

 #[cfg(test)]
 mod tests {
-
     use arrow_array::Array;

     use super::*;

There were several other places in the decoder where I wasn't sure if the reserve function was being called appropriately which should also be studied

[alamb]

I spent some time with @lyang24 working on this issue with #9093

I printed out the buffer capacity for reads. They are almost all 0 and the reader doesn't then reserve enough space

You can actually see most of the reads have an empty buffer

ewArrayDecoder::read called with len 8192, current views capacity: 0
ByteViewArrayDecoder::read called with len 8192, current views capacity: 0
ByteViewArrayDecoder::read called with len 8192, current views capacity: 0
ByteViewArrayDecoder::read called with len 8192, current views capacity: 0
ByteViewArrayDecoder::read called with len 8192, current views capacity: 0
ByteViewArrayDecoder::read called with len 8192, current views capacity: 0
ByteViewArrayDecoder::read called with len 8192, current views capacity: 0
Read batch with 8192 rows and 105 columns
ByteViewArrayDecoder::read called with len 5120, current views capacity: 0
ByteViewArrayDecoder::read called with len 3072, current views capacity: 5120
ByteViewArrayDecoder::read called with len 6144, current views capacity: 0
ByteViewArrayDecoder::read called with len 2048, current views capacity: 6144

I tracked it down in a debugger and the default buffer is being created here:

https://github.com/apache/arrow-rs/blob/02fa779a9cb122c5218293be3afb980832701683/parquet/src/arrow/record_reader/mod.rs#L76-L75

ByteViewArrayDecoderPlain does reserve capacity https://github.com/apache/arrow-rs/blob/e2b2b8f5ec9ccf10bf7d584cf94e56018e2ac800/parquet/src/arrow/array_reader/byte_view_array.rs#L333-L332

However, ByteViewArrayDecoderDictionary::read does not seem to reserve capacity
https://github.com/apache/arrow-rs/blob/e2b2b8f5ec9ccf10bf7d584cf94e56018e2ac800/parquet/src/arrow/array_reader/byte_view_array.rs#L442-L441

Whole Test Program

use std::fs::File;
use std::io::{BufReader, Read};
use std::sync::Arc;
use arrow::datatypes::{DataType, Field, FieldRef, Schema};
use parquet::arrow::arrow_reader::{ArrowReaderOptions, ParquetRecordBatchReaderBuilder};
use bytes::Bytes;
use parquet::file::metadata::ParquetMetaDataReader;

fn main() {
    let file_name = "/Users/andrewlamb/Downloads/hits/hits_0.parquet";
    println!("Opening file: {file_name}", );
    let mut file = File::open(file_name).unwrap();
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes).unwrap();
    let bytes = Bytes::from(bytes);

    let schema = string_to_view_types(ParquetRecordBatchReaderBuilder::try_new(bytes.clone()).unwrap().schema());

    //println!("Schema: {:?}", schema);

    let options = ArrowReaderOptions::new()
        .with_schema(schema);
    let reader = ParquetRecordBatchReaderBuilder::try_new_with_options(bytes, options).unwrap()
        .with_batch_size(8192)
        .build().unwrap();

    for batch in reader {
        let batch = batch.unwrap();
        println!("Read batch with {} rows and {} columns", batch.num_rows(), batch.num_columns());
    }

    println!("Done");


}


// Hack because the clickbench files were written with the wrong logical type for strings
fn string_to_view_types(schema: &Arc<Schema>) -> Arc<Schema> {
    let fields: Vec<FieldRef> = schema
        .fields()
        .iter()
        .map(|field| {
            let existing_type = field.data_type();
            if existing_type == &DataType::Utf8 || existing_type == &DataType::Binary {
                Arc::new(Field::new(
                    field.name(),
                    DataType::Utf8View,
                    field.is_nullable(),
                ))
            } else {
                Arc::clone(field)
            }
        })
        .collect();
    Arc::new(Schema::new(fields))
}

[alamb]

I think we could probably sprinkly some "reserves" in there locally, but I think we can probably do much better if we reserve the correct capacity when creating the output buffers in the first place (as we know the (max) output size = batch_size)

i have some ideas on how to wire this in.

Basically, it will involve https://github.com/apache/arrow-rs/blob/9a1e8b572d11078e813fffe3d5c7106b6953d58c/parquet/src/arrow/record_reader/buffer.rs#L21-L20

Changing ValuesBuffer so that instead of

pub trait ValuesBuffer : default {
...
}

It must be explicitly created with capacity

pub trait ValuesBuffer {
  /// create a new buffer with an allocation  for `capacity` items
  fn new_with_capacity(capacity: usize) -> Self;
}

And then use the compiler to find all the relevant places and add the appropriate capacies