From f53bf5de87379b1bc1ba0708d5e1a3f8cbf80c0a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 12:41:04 +0100
Subject: [PATCH 01/16] Eliminate per-element dict bounds checks in RLE
 bit-packed path

When bit_width guarantees all possible indices fit within the dictionary,
use unchecked indexing to allow LLVM to unroll the dict gather loop 4x
with paired loads/stores instead of scalar with per-element bounds checks.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/encodings/rle.rs | 23 +++++++++++++++++++----
 1 file changed, 19 insertions(+), 4 deletions(-)

diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index c95a46c634d2..18a94eef3ae8 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -458,6 +458,11 @@ impl RleDecoder {
     {
         assert!(buffer.len() >= max_values);
 
+        // If all possible bit_width-bit values are valid dict indices,
+        // we can skip per-element bounds checks in the hot loop.
+        let all_indices_valid = self.bit_width < 64
+            && dict.len() >= (1usize << self.bit_width);
+
         let mut values_read = 0;
         while values_read < max_values {
             let index_buf = self.index_buf.get_or_insert_with(|| Box::new([0; 1024]));
@@ -494,10 +499,20 @@ impl RleDecoder {
                         self.bit_packed_left = 0;
                         break;
                     }
-                    buffer[values_read..values_read + num_values]
-                        .iter_mut()
-                        .zip(index_buf[..num_values].iter())
-                        .for_each(|(b, i)| b.clone_from(&dict[*i as usize]));
+                    if all_indices_valid {
+                        // SAFETY: bit_width guarantees all decoded indices are < dict.len()
+                        buffer[values_read..values_read + num_values]
+                            .iter_mut()
+                            .zip(index_buf[..num_values].iter())
+                            .for_each(|(b, i)| unsafe {
+                                b.clone_from(dict.get_unchecked(*i as usize));
+                            });
+                    } else {
+                        buffer[values_read..values_read + num_values]
+                            .iter_mut()
+                            .zip(index_buf[..num_values].iter())
+                            .for_each(|(b, i)| b.clone_from(&dict[*i as usize]));
+                    }
                     self.bit_packed_left -= num_values as u32;
                     values_read += num_values;
                     if num_values < to_read {

From b53b5660a4a1346076bdd215ea44c7c95f08f9a4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 20:56:12 +0100
Subject: [PATCH 02/16] Add get_batch_direct method to RleDecoder

Add RleDecodedBatch enum and get_batch_direct method that exposes RLE vs
bit-packed batches via callback, allowing callers to handle each case
optimally without going through the index buffer.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/encodings/rle.rs | 54 ++++++++++++++++++++++++++++++++++++
 1 file changed, 54 insertions(+)

diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index 18a94eef3ae8..da5e326b109d 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -287,6 +287,14 @@ impl RleEncoder {
 /// Size, in number of `i32s` of buffer to use for RLE batch reading
 const RLE_DECODER_INDEX_BUFFER_SIZE: usize = 1024;
 
+/// A decoded batch from [`RleDecoder::get_batch_direct`].
+pub enum RleDecodedBatch<'a> {
+    /// An RLE run: all values are the same index, repeated `count` times
+    Rle { index: i32, count: usize },
+    /// A batch of bit-packed indices
+    BitPacked(&'a [i32]),
+}
+
 /// A RLE/Bit-Packing hybrid decoder.
 pub struct RleDecoder {
     // Number of bits used to encode the value. Must be between [0, 64].
@@ -414,6 +422,52 @@ impl RleDecoder {
         Ok(values_read)
     }
 
+    /// Decode up to `max_values` indices and call `f` with each decoded batch.
+    ///
+    /// For RLE runs, provides [`RleDecodedBatch::Rle`] so callers can fill output directly.
+    /// For bit-packed runs, provides [`RleDecodedBatch::BitPacked`] with decoded indices.
+    pub fn get_batch_direct<F>(
+        &mut self,
+        max_values: usize,
+        mut f: F,
+    ) -> Result<usize>
+    where
+        F: FnMut(RleDecodedBatch<'_>),
+    {
+        let mut values_read = 0;
+        let mut index_buf = [0i32; 1024];
+        while values_read < max_values {
+            if self.rle_left > 0 {
+                let num_values = cmp::min(max_values - values_read, self.rle_left as usize);
+                let idx = self.current_value.unwrap() as i32;
+                f(RleDecodedBatch::Rle { index: idx, count: num_values });
+                self.rle_left -= num_values as u32;
+                values_read += num_values;
+            } else if self.bit_packed_left > 0 {
+                let to_read = (max_values - values_read)
+                    .min(self.bit_packed_left as usize)
+                    .min(index_buf.len());
+                let bit_reader = self
+                    .bit_reader
+                    .as_mut()
+                    .ok_or_else(|| general_err!("bit_reader should be set"))?;
+
+                let num_values =
+                    bit_reader.get_batch::<i32>(&mut index_buf[..to_read], self.bit_width as usize);
+                if num_values == 0 {
+                    self.bit_packed_left = 0;
+                    continue;
+                }
+                f(RleDecodedBatch::BitPacked(&index_buf[..num_values]));
+                self.bit_packed_left -= num_values as u32;
+                values_read += num_values;
+            } else if !self.reload()? {
+                break;
+            }
+        }
+        Ok(values_read)
+    }
+
     #[inline(never)]
     pub fn skip(&mut self, num_values: usize) -> Result<usize> {
         let mut values_skipped = 0;

From d00db2f78e7f23244d2159f954c6b53496ad7966 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 20:56:22 +0100
Subject: [PATCH 03/16] Use branchless index clamping in get_batch_with_dict

Replace if/else checked/unchecked branching with a single branchless
.min(max_idx) clamp. This prevents UB on corrupt parquet files while
avoiding per-element bounds checks.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/encodings/rle.rs | 30 ++++++++++++------------------
 1 file changed, 12 insertions(+), 18 deletions(-)

diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index da5e326b109d..0b62fbee59be 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -512,10 +512,12 @@ impl RleDecoder {
     {
         assert!(buffer.len() >= max_values);
 
-        // If all possible bit_width-bit values are valid dict indices,
-        // we can skip per-element bounds checks in the hot loop.
-        let all_indices_valid = self.bit_width < 64
-            && dict.len() >= (1usize << self.bit_width);
+        if dict.is_empty() {
+            return Ok(0);
+        }
+        // Clamp index to valid range to prevent UB on corrupt data.
+        // This is branchless (cmp+csel on ARM) and avoids bounds checks in the hot loop.
+        let max_idx = dict.len() - 1;
 
         let mut values_read = 0;
         while values_read < max_values {
@@ -553,20 +555,12 @@ impl RleDecoder {
                         self.bit_packed_left = 0;
                         break;
                     }
-                    if all_indices_valid {
-                        // SAFETY: bit_width guarantees all decoded indices are < dict.len()
-                        buffer[values_read..values_read + num_values]
-                            .iter_mut()
-                            .zip(index_buf[..num_values].iter())
-                            .for_each(|(b, i)| unsafe {
-                                b.clone_from(dict.get_unchecked(*i as usize));
-                            });
-                    } else {
-                        buffer[values_read..values_read + num_values]
-                            .iter_mut()
-                            .zip(index_buf[..num_values].iter())
-                            .for_each(|(b, i)| b.clone_from(&dict[*i as usize]));
-                    }
+                    buffer[values_read..values_read + num_values]
+                        .iter_mut()
+                        .zip(index_buf[..num_values].iter())
+                        .for_each(|(b, i)| unsafe {
+                            b.clone_from(dict.get_unchecked((*i as usize).min(max_idx)));
+                        });
                     self.bit_packed_left -= num_values as u32;
                     values_read += num_values;
                     if num_values < to_read {

From 01320dd725fceca315784523d9db5a03ae1e951d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 18:31:36 +0100
Subject: [PATCH 04/16] Gate RleDecodedBatch and get_batch_direct behind arrow
 feature

These are only used by the arrow dictionary_index decoder. Without
the arrow feature, they appear as dead code to clippy.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/encodings/rle.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index 0b62fbee59be..bee4d00d4337 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -288,6 +288,7 @@ impl RleEncoder {
 const RLE_DECODER_INDEX_BUFFER_SIZE: usize = 1024;
 
 /// A decoded batch from [`RleDecoder::get_batch_direct`].
+#[cfg(feature = "arrow")]
 pub enum RleDecodedBatch<'a> {
     /// An RLE run: all values are the same index, repeated `count` times
     Rle { index: i32, count: usize },
@@ -426,6 +427,7 @@ impl RleDecoder {
     ///
     /// For RLE runs, provides [`RleDecodedBatch::Rle`] so callers can fill output directly.
     /// For bit-packed runs, provides [`RleDecodedBatch::BitPacked`] with decoded indices.
+    #[cfg(feature = "arrow")]
     pub fn get_batch_direct<F>(
         &mut self,
         max_values: usize,

From cff5bde65ecb481bcf4ee20e7d75bc46aabe2442 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 13:55:51 +0100
Subject: [PATCH 05/16] Optimize StringView dictionary decoding with unchecked
 index access

When bit_width guarantees all possible indices fit within the dictionary,
use unchecked access to eliminate per-element bounds checks. Also skip
buffer management when all dictionary views are inlined (<=12 bytes).

Generates a clean 8-instruction gather loop for the common case
(all_indices_valid + base_buffer_idx=0) and a branchless 14-instruction
loop for the non-zero buffer offset case.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 .../src/arrow/array_reader/byte_view_array.rs | 81 +++++++++++++------
 parquet/src/arrow/decoder/dictionary_index.rs | 10 +++
 2 files changed, 66 insertions(+), 25 deletions(-)

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 65b627aae451..fe9ac0b0188f 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -475,38 +475,70 @@ impl ByteViewArrayDecoderDictionary {
             return Ok(0);
         }
 
-        // Check if the last few buffer of `output`` are the same as the `dict` buffer
-        // This is to avoid creating a new buffers if the same dictionary is used for multiple `read`
-        let need_to_create_new_buffer = {
-            if output.buffers.len() >= dict.buffers.len() {
-                let offset = output.buffers.len() - dict.buffers.len();
-                output.buffers[offset..]
-                    .iter()
-                    .zip(dict.buffers.iter())
-                    .any(|(a, b)| !a.ptr_eq(b))
-            } else {
-                true
-            }
-        };
+        // Check if all dictionary views are inlined (len <= 12).
+        // If so, we can skip buffer management entirely since inline views
+        // don't reference any buffers.
+        let all_views_inlined = dict.views.iter().all(|&v| (v as u32) <= 12);
+
+        if !all_views_inlined {
+            // Check if the last few buffer of `output`` are the same as the `dict` buffer
+            // This is to avoid creating a new buffers if the same dictionary is used for multiple `read`
+            let need_to_create_new_buffer = {
+                if output.buffers.len() >= dict.buffers.len() {
+                    let offset = output.buffers.len() - dict.buffers.len();
+                    output.buffers[offset..]
+                        .iter()
+                        .zip(dict.buffers.iter())
+                        .any(|(a, b)| !a.ptr_eq(b))
+                } else {
+                    true
+                }
+            };
 
-        if need_to_create_new_buffer {
-            for b in dict.buffers.iter() {
-                output.buffers.push(b.clone());
+            if need_to_create_new_buffer {
+                for b in dict.buffers.iter() {
+                    output.buffers.push(b.clone());
+                }
             }
         }
 
         // Calculate the offset of the dictionary buffers in the output buffers
-        // For example if the 2nd buffer in the dictionary is the 5th buffer in the output buffers,
-        // then the base_buffer_idx is 5 - 2 = 3
-        let base_buffer_idx = output.buffers.len() as u32 - dict.buffers.len() as u32;
+        let base_buffer_idx = if all_views_inlined {
+            0
+        } else {
+            output.buffers.len() as u32 - dict.buffers.len() as u32
+        };
+
+        let dict_views = &dict.views;
+        // If bit_width guarantees all indices are valid, skip per-element bounds checks
+        let all_indices_valid = self.decoder.all_indices_valid_for(dict_views.len());
 
         let mut error = None;
         let read = self.decoder.read(len, |keys| {
-            if base_buffer_idx == 0 {
-                // the dictionary buffers are the last buffers in output, we can directly use the views
+            if all_indices_valid {
+                if base_buffer_idx == 0 {
+                    // SAFETY: bit_width guarantees all decoded indices < dict_views.len()
+                    output.views.extend(keys.iter().map(|k| unsafe {
+                        *dict_views.get_unchecked(*k as usize)
+                    }));
+                } else {
+                    // SAFETY: bit_width guarantees all decoded indices < dict_views.len()
+                    output.views.extend(keys.iter().map(|k| {
+                        let view = unsafe { *dict_views.get_unchecked(*k as usize) };
+                        let len = view as u32;
+                        if len <= 12 {
+                            view
+                        } else {
+                            let mut view = ByteView::from(view);
+                            view.buffer_index += base_buffer_idx;
+                            view.into()
+                        }
+                    }));
+                }
+            } else if base_buffer_idx == 0 {
                 output
                     .views
-                    .extend(keys.iter().map(|k| match dict.views.get(*k as usize) {
+                    .extend(keys.iter().map(|k| match dict_views.get(*k as usize) {
                         Some(&view) => view,
                         None => {
                             if error.is_none() {
@@ -515,11 +547,10 @@ impl ByteViewArrayDecoderDictionary {
                             0
                         }
                     }));
-                Ok(())
             } else {
                 output
                     .views
-                    .extend(keys.iter().map(|k| match dict.views.get(*k as usize) {
+                    .extend(keys.iter().map(|k| match dict_views.get(*k as usize) {
                         Some(&view) => {
                             let len = view as u32;
                             if len <= 12 {
@@ -537,8 +568,8 @@ impl ByteViewArrayDecoderDictionary {
                             0
                         }
                     }));
-                Ok(())
             }
+            Ok(())
         })?;
         if let Some(e) = error {
             return Err(e);
diff --git a/parquet/src/arrow/decoder/dictionary_index.rs b/parquet/src/arrow/decoder/dictionary_index.rs
index 7a4b77f89d59..fa52bcbbc8f9 100644
--- a/parquet/src/arrow/decoder/dictionary_index.rs
+++ b/parquet/src/arrow/decoder/dictionary_index.rs
@@ -25,6 +25,9 @@ pub struct DictIndexDecoder {
     /// Decoder for the dictionary offsets array
     decoder: RleDecoder,
 
+    /// The bit width used to encode dictionary indices
+    bit_width: u8,
+
     /// We want to decode the offsets in chunks so we will maintain an internal buffer of decoded
     /// offsets
     index_buf: Box<[i32; 1024]>,
@@ -49,6 +52,7 @@ impl DictIndexDecoder {
 
         Ok(Self {
             decoder,
+            bit_width,
             index_buf: Box::new([0; 1024]),
             index_buf_len: 0,
             index_offset: 0,
@@ -56,6 +60,12 @@ impl DictIndexDecoder {
         })
     }
 
+    /// Returns true if all possible indices that could be encoded with `bit_width`
+    /// bits are valid for a dictionary of the given length.
+    pub fn all_indices_valid_for(&self, dict_len: usize) -> bool {
+        self.bit_width < 64 && dict_len >= (1usize << self.bit_width)
+    }
+
     /// Read up to `len` values, returning the number of values read
     /// and calling `f` with each decoded dictionary index
     ///

From 937082810ccf63d838b1464612b3451fe49acfa7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 14:50:12 +0100
Subject: [PATCH 06/16] Pre-reserve output views in StringView dictionary
 decoding

Reserve the full output capacity upfront before the decode loop,
eliminating per-chunk reallocation checks inside extend. This gives
a ~25% speedup for dictionary-encoded StringView reads.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/arrow/array_reader/byte_view_array.rs | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index fe9ac0b0188f..9c6178b5f30d 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -513,6 +513,9 @@ impl ByteViewArrayDecoderDictionary {
         // If bit_width guarantees all indices are valid, skip per-element bounds checks
         let all_indices_valid = self.decoder.all_indices_valid_for(dict_views.len());
 
+        // Pre-reserve output capacity to avoid per-chunk reallocation in extend
+        output.views.reserve(len);
+
         let mut error = None;
         let read = self.decoder.read(len, |keys| {
             if all_indices_valid {

From 54e39c25452211c7d79b9651660f6a7b369c62ac Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 20:57:04 +0100
Subject: [PATCH 07/16] Fuse RLE decoding and dict view gathering for
 StringView

For RLE runs, look up the dict view once and repeat directly with
repeat_n, skipping the index buffer entirely. For bit-packed runs,
decode indices to a stack-local buffer and gather immediately.

Skip buffer management when all dictionary views are inlined (<=12 bytes).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 .../src/arrow/array_reader/byte_view_array.rs | 72 ++------------
 parquet/src/arrow/decoder/dictionary_index.rs | 96 ++++++++++++++++---
 2 files changed, 88 insertions(+), 80 deletions(-)

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 9c6178b5f30d..c94f6d6fda90 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -30,7 +30,6 @@ use crate::schema::types::ColumnDescPtr;
 use crate::util::utf8::check_valid_utf8;
 use arrow_array::{ArrayRef, builder::make_view};
 use arrow_buffer::Buffer;
-use arrow_data::ByteView;
 use arrow_schema::DataType as ArrowType;
 use bytes::Bytes;
 use std::any::Any;
@@ -509,74 +508,15 @@ impl ByteViewArrayDecoderDictionary {
             output.buffers.len() as u32 - dict.buffers.len() as u32
         };
 
-        let dict_views = &dict.views;
-        // If bit_width guarantees all indices are valid, skip per-element bounds checks
-        let all_indices_valid = self.decoder.all_indices_valid_for(dict_views.len());
-
         // Pre-reserve output capacity to avoid per-chunk reallocation in extend
         output.views.reserve(len);
 
-        let mut error = None;
-        let read = self.decoder.read(len, |keys| {
-            if all_indices_valid {
-                if base_buffer_idx == 0 {
-                    // SAFETY: bit_width guarantees all decoded indices < dict_views.len()
-                    output.views.extend(keys.iter().map(|k| unsafe {
-                        *dict_views.get_unchecked(*k as usize)
-                    }));
-                } else {
-                    // SAFETY: bit_width guarantees all decoded indices < dict_views.len()
-                    output.views.extend(keys.iter().map(|k| {
-                        let view = unsafe { *dict_views.get_unchecked(*k as usize) };
-                        let len = view as u32;
-                        if len <= 12 {
-                            view
-                        } else {
-                            let mut view = ByteView::from(view);
-                            view.buffer_index += base_buffer_idx;
-                            view.into()
-                        }
-                    }));
-                }
-            } else if base_buffer_idx == 0 {
-                output
-                    .views
-                    .extend(keys.iter().map(|k| match dict_views.get(*k as usize) {
-                        Some(&view) => view,
-                        None => {
-                            if error.is_none() {
-                                error = Some(general_err!("invalid key={} for dictionary", *k));
-                            }
-                            0
-                        }
-                    }));
-            } else {
-                output
-                    .views
-                    .extend(keys.iter().map(|k| match dict_views.get(*k as usize) {
-                        Some(&view) => {
-                            let len = view as u32;
-                            if len <= 12 {
-                                view
-                            } else {
-                                let mut view = ByteView::from(view);
-                                view.buffer_index += base_buffer_idx;
-                                view.into()
-                            }
-                        }
-                        None => {
-                            if error.is_none() {
-                                error = Some(general_err!("invalid key={} for dictionary", *k));
-                            }
-                            0
-                        }
-                    }));
-            }
-            Ok(())
-        })?;
-        if let Some(e) = error {
-            return Err(e);
-        }
+        let read = self.decoder.read_gather_views(
+            len,
+            &dict.views,
+            &mut output.views,
+            base_buffer_idx,
+        )?;
         Ok(read)
     }
 
diff --git a/parquet/src/arrow/decoder/dictionary_index.rs b/parquet/src/arrow/decoder/dictionary_index.rs
index fa52bcbbc8f9..cc4aac2667ef 100644
--- a/parquet/src/arrow/decoder/dictionary_index.rs
+++ b/parquet/src/arrow/decoder/dictionary_index.rs
@@ -17,7 +17,7 @@
 
 use bytes::Bytes;
 
-use crate::encodings::rle::RleDecoder;
+use crate::encodings::rle::{RleDecodedBatch, RleDecoder};
 use crate::errors::Result;
 
 /// Decoder for `Encoding::RLE_DICTIONARY` indices
@@ -25,12 +25,9 @@ pub struct DictIndexDecoder {
     /// Decoder for the dictionary offsets array
     decoder: RleDecoder,
 
-    /// The bit width used to encode dictionary indices
-    bit_width: u8,
-
     /// We want to decode the offsets in chunks so we will maintain an internal buffer of decoded
     /// offsets
-    index_buf: Box<[i32; 1024]>,
+    index_buf: Vec<i32>,
     /// Current length of `index_buf`
     index_buf_len: usize,
     /// Current offset into `index_buf`. If `index_buf_offset` == `index_buf_len` then we've consumed
@@ -49,23 +46,17 @@ impl DictIndexDecoder {
         let bit_width = data[0];
         let mut decoder = RleDecoder::new(bit_width);
         decoder.set_data(data.slice(1..))?;
+        let max_remaining = num_values.unwrap_or(num_levels);
 
         Ok(Self {
             decoder,
-            bit_width,
-            index_buf: Box::new([0; 1024]),
+            index_buf: vec![0; max_remaining.min(1024)],
             index_buf_len: 0,
             index_offset: 0,
-            max_remaining_values: num_values.unwrap_or(num_levels),
+            max_remaining_values: max_remaining,
         })
     }
 
-    /// Returns true if all possible indices that could be encoded with `bit_width`
-    /// bits are valid for a dictionary of the given length.
-    pub fn all_indices_valid_for(&self, dict_len: usize) -> bool {
-        self.bit_width < 64 && dict_len >= (1usize << self.bit_width)
-    }
-
     /// Read up to `len` values, returning the number of values read
     /// and calling `f` with each decoded dictionary index
     ///
@@ -101,6 +92,83 @@ impl DictIndexDecoder {
         Ok(values_read)
     }
 
+    /// Decode indices and gather views directly into `output`.
+    ///
+    /// For RLE runs, fills output with the repeated view directly (no index buffer needed).
+    /// For bit-packed runs, decodes indices to a stack-local buffer and gathers immediately.
+    pub fn read_gather_views(
+        &mut self,
+        len: usize,
+        dict_views: &[u128],
+        output: &mut Vec<u128>,
+        base_buffer_idx: u32,
+    ) -> Result<usize> {
+        let to_read = len.min(self.max_remaining_values);
+        let mut values_read = 0;
+
+        // Flush any buffered indices from previous reads
+        if self.index_offset < self.index_buf_len {
+            let n = (self.index_buf_len - self.index_offset).min(to_read);
+            let keys = &self.index_buf[self.index_offset..self.index_offset + n];
+            Self::gather_views(keys, dict_views, output, base_buffer_idx);
+            self.index_offset += n;
+            self.max_remaining_values -= n;
+            values_read += n;
+        }
+
+        if values_read < to_read {
+            let read = self.decoder.get_batch_direct(to_read - values_read, |batch| {
+                match batch {
+                    RleDecodedBatch::Rle { index, count } => {
+                        // SAFETY: RLE value was decoded from the stream; validated by caller
+                        let view = unsafe { *dict_views.get_unchecked(index as usize) };
+                        let view = if base_buffer_idx == 0 || (view as u32) <= 12 {
+                            view
+                        } else {
+                            let mut bv = arrow_data::ByteView::from(view);
+                            bv.buffer_index += base_buffer_idx;
+                            bv.into()
+                        };
+                        output.extend(std::iter::repeat_n(view, count));
+                    }
+                    RleDecodedBatch::BitPacked(keys) => {
+                        Self::gather_views(keys, dict_views, output, base_buffer_idx);
+                    }
+                }
+            })?;
+            self.max_remaining_values -= read;
+            values_read += read;
+        }
+
+        Ok(values_read)
+    }
+
+    fn gather_views(
+        keys: &[i32],
+        dict_views: &[u128],
+        output: &mut Vec<u128>,
+        base_buffer_idx: u32,
+    ) {
+        if base_buffer_idx == 0 {
+            // SAFETY: caller ensures all indices are valid
+            output.extend(keys.iter().map(|k| unsafe {
+                *dict_views.get_unchecked(*k as usize)
+            }));
+        } else {
+            // SAFETY: caller ensures all indices are valid
+            output.extend(keys.iter().map(|k| {
+                let view = unsafe { *dict_views.get_unchecked(*k as usize) };
+                if (view as u32) <= 12 {
+                    view
+                } else {
+                    let mut bv = arrow_data::ByteView::from(view);
+                    bv.buffer_index += base_buffer_idx;
+                    bv.into()
+                }
+            }));
+        }
+    }
+
     /// Skip up to `to_skip` values, returning the number of values skipped
     pub fn skip(&mut self, to_skip: usize) -> Result<usize> {
         let to_skip = to_skip.min(self.max_remaining_values);

From 5e3e6a7fd0ba0d4a5b4fed71068458dffe095ed0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 20:57:24 +0100
Subject: [PATCH 08/16] Use branchless index clamping in view gather paths

Apply .min(max_idx) clamping in gather_views to prevent UB on corrupt
data while keeping the hot loop branchless.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/arrow/decoder/dictionary_index.rs | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/parquet/src/arrow/decoder/dictionary_index.rs b/parquet/src/arrow/decoder/dictionary_index.rs
index cc4aac2667ef..6bd1da72c133 100644
--- a/parquet/src/arrow/decoder/dictionary_index.rs
+++ b/parquet/src/arrow/decoder/dictionary_index.rs
@@ -120,8 +120,7 @@ impl DictIndexDecoder {
             let read = self.decoder.get_batch_direct(to_read - values_read, |batch| {
                 match batch {
                     RleDecodedBatch::Rle { index, count } => {
-                        // SAFETY: RLE value was decoded from the stream; validated by caller
-                        let view = unsafe { *dict_views.get_unchecked(index as usize) };
+                        let view = dict_views[index as usize];
                         let view = if base_buffer_idx == 0 || (view as u32) <= 12 {
                             view
                         } else {
@@ -149,15 +148,16 @@ impl DictIndexDecoder {
         output: &mut Vec<u128>,
         base_buffer_idx: u32,
     ) {
+        // Clamp index to valid range to prevent UB on corrupt data.
+        // This is branchless (cmp+csel on ARM) and avoids bounds checks in the hot loop.
+        let max_idx = dict_views.len() - 1;
         if base_buffer_idx == 0 {
-            // SAFETY: caller ensures all indices are valid
             output.extend(keys.iter().map(|k| unsafe {
-                *dict_views.get_unchecked(*k as usize)
+                *dict_views.get_unchecked((*k as usize).min(max_idx))
             }));
         } else {
-            // SAFETY: caller ensures all indices are valid
             output.extend(keys.iter().map(|k| {
-                let view = unsafe { *dict_views.get_unchecked(*k as usize) };
+                let view = unsafe { *dict_views.get_unchecked((*k as usize).min(max_idx)) };
                 if (view as u32) <= 12 {
                     view
                 } else {

From cdd3ea40aac5a2e25bc266a1a323cf15195ad969 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 18:21:07 +0100
Subject: [PATCH 09/16] Pre-reserve offsets in ByteArray dictionary decoding

Reserve offsets capacity upfront before the decode loop to avoid
per-chunk reallocation. ~3.5% improvement for StringArray dict reads.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/arrow/array_reader/byte_array.rs | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/parquet/src/arrow/array_reader/byte_array.rs b/parquet/src/arrow/array_reader/byte_array.rs
index 0acbe6501924..2d0d44fbe203 100644
--- a/parquet/src/arrow/array_reader/byte_array.rs
+++ b/parquet/src/arrow/array_reader/byte_array.rs
@@ -580,6 +580,9 @@ impl ByteArrayDecoderDictionary {
             return Ok(0);
         }
 
+        // Pre-reserve offsets capacity to avoid per-chunk reallocation
+        output.offsets.reserve(len);
+
         self.decoder.read(len, |keys| {
             output.extend_from_dictionary(keys, dict.offsets.as_slice(), dict.values.as_slice())
         })

From 6e800061298dccdf341795bc305d0deca19d93e4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Thu, 19 Mar 2026 21:19:23 +0100
Subject: [PATCH 10/16] fmt

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/arrow/decoder/dictionary_index.rs | 43 ++++++++++---------
 1 file changed, 22 insertions(+), 21 deletions(-)

diff --git a/parquet/src/arrow/decoder/dictionary_index.rs b/parquet/src/arrow/decoder/dictionary_index.rs
index 6bd1da72c133..ba16c5687560 100644
--- a/parquet/src/arrow/decoder/dictionary_index.rs
+++ b/parquet/src/arrow/decoder/dictionary_index.rs
@@ -117,24 +117,24 @@ impl DictIndexDecoder {
         }
 
         if values_read < to_read {
-            let read = self.decoder.get_batch_direct(to_read - values_read, |batch| {
-                match batch {
-                    RleDecodedBatch::Rle { index, count } => {
-                        let view = dict_views[index as usize];
-                        let view = if base_buffer_idx == 0 || (view as u32) <= 12 {
-                            view
-                        } else {
-                            let mut bv = arrow_data::ByteView::from(view);
-                            bv.buffer_index += base_buffer_idx;
-                            bv.into()
-                        };
-                        output.extend(std::iter::repeat_n(view, count));
-                    }
-                    RleDecodedBatch::BitPacked(keys) => {
-                        Self::gather_views(keys, dict_views, output, base_buffer_idx);
-                    }
-                }
-            })?;
+            let read =
+                self.decoder
+                    .get_batch_direct(to_read - values_read, |batch| match batch {
+                        RleDecodedBatch::Rle { index, count } => {
+                            let view = dict_views[index as usize];
+                            let view = if base_buffer_idx == 0 || (view as u32) <= 12 {
+                                view
+                            } else {
+                                let mut bv = arrow_data::ByteView::from(view);
+                                bv.buffer_index += base_buffer_idx;
+                                bv.into()
+                            };
+                            output.extend(std::iter::repeat_n(view, count));
+                        }
+                        RleDecodedBatch::BitPacked(keys) => {
+                            Self::gather_views(keys, dict_views, output, base_buffer_idx);
+                        }
+                    })?;
             self.max_remaining_values -= read;
             values_read += read;
         }
@@ -152,9 +152,10 @@ impl DictIndexDecoder {
         // This is branchless (cmp+csel on ARM) and avoids bounds checks in the hot loop.
         let max_idx = dict_views.len() - 1;
         if base_buffer_idx == 0 {
-            output.extend(keys.iter().map(|k| unsafe {
-                *dict_views.get_unchecked((*k as usize).min(max_idx))
-            }));
+            output.extend(
+                keys.iter()
+                    .map(|k| unsafe { *dict_views.get_unchecked((*k as usize).min(max_idx)) }),
+            );
         } else {
             output.extend(keys.iter().map(|k| {
                 let view = unsafe { *dict_views.get_unchecked((*k as usize).min(max_idx)) };

From 360a6726387410b94e2f9bd31f6b0fe0d7e8411a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Fri, 20 Mar 2026 06:27:58 +0100
Subject: [PATCH 11/16] Format cfg(feature = "arrow") gated code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/arrow/array_reader/byte_view_array.rs |  9 +++------
 parquet/src/encodings/rle.rs                      | 11 +++++------
 2 files changed, 8 insertions(+), 12 deletions(-)

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index c94f6d6fda90..be6f600603df 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -511,12 +511,9 @@ impl ByteViewArrayDecoderDictionary {
         // Pre-reserve output capacity to avoid per-chunk reallocation in extend
         output.views.reserve(len);
 
-        let read = self.decoder.read_gather_views(
-            len,
-            &dict.views,
-            &mut output.views,
-            base_buffer_idx,
-        )?;
+        let read =
+            self.decoder
+                .read_gather_views(len, &dict.views, &mut output.views, base_buffer_idx)?;
         Ok(read)
     }
 
diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index bee4d00d4337..5f11cc51fcb8 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -428,11 +428,7 @@ impl RleDecoder {
     /// For RLE runs, provides [`RleDecodedBatch::Rle`] so callers can fill output directly.
     /// For bit-packed runs, provides [`RleDecodedBatch::BitPacked`] with decoded indices.
     #[cfg(feature = "arrow")]
-    pub fn get_batch_direct<F>(
-        &mut self,
-        max_values: usize,
-        mut f: F,
-    ) -> Result<usize>
+    pub fn get_batch_direct<F>(&mut self, max_values: usize, mut f: F) -> Result<usize>
     where
         F: FnMut(RleDecodedBatch<'_>),
     {
@@ -442,7 +438,10 @@ impl RleDecoder {
             if self.rle_left > 0 {
                 let num_values = cmp::min(max_values - values_read, self.rle_left as usize);
                 let idx = self.current_value.unwrap() as i32;
-                f(RleDecodedBatch::Rle { index: idx, count: num_values });
+                f(RleDecodedBatch::Rle {
+                    index: idx,
+                    count: num_values,
+                });
                 self.rle_left -= num_values as u32;
                 values_read += num_values;
             } else if self.bit_packed_left > 0 {

From 100c5dea740444b8dce4973673f55cb0cd585279 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Wed, 25 Mar 2026 08:53:03 +0100
Subject: [PATCH 12/16] Cache all_views_inlined on ViewBuffer, restore error
 handling in gather_views

Precompute all_views_inlined when the dictionary is set instead of
recomputing it on every read call. Replace branchless index clamping
with proper bounds checking using the deferred error pattern to keep
the extend loop fast.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 .../src/arrow/array_reader/byte_view_array.rs |  4 +-
 parquet/src/arrow/buffer/view_buffer.rs       | 15 ++++
 parquet/src/arrow/decoder/dictionary_index.rs | 88 +++++++++++++------
 3 files changed, 80 insertions(+), 27 deletions(-)

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index be6f600603df..cb27b85bfd11 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -169,6 +169,7 @@ impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
             self.validate_utf8,
         );
         decoder.read(&mut buffer, usize::MAX)?;
+        buffer.compute_all_views_inlined();
         self.dict = Some(buffer);
         Ok(())
     }
@@ -477,7 +478,8 @@ impl ByteViewArrayDecoderDictionary {
         // Check if all dictionary views are inlined (len <= 12).
         // If so, we can skip buffer management entirely since inline views
         // don't reference any buffers.
-        let all_views_inlined = dict.views.iter().all(|&v| (v as u32) <= 12);
+        // This value is precomputed when the dictionary is set.
+        let all_views_inlined = dict.all_views_inlined().unwrap_or(false);
 
         if !all_views_inlined {
             // Check if the last few buffer of `output`` are the same as the `dict` buffer
diff --git a/parquet/src/arrow/buffer/view_buffer.rs b/parquet/src/arrow/buffer/view_buffer.rs
index a93674663f7b..1e73b86ce0b3 100644
--- a/parquet/src/arrow/buffer/view_buffer.rs
+++ b/parquet/src/arrow/buffer/view_buffer.rs
@@ -30,6 +30,9 @@ use std::sync::Arc;
 pub struct ViewBuffer {
     pub views: Vec<u128>,
     pub buffers: Vec<Buffer>,
+    /// Whether all views are inlined (length <= 12), meaning no buffer references are needed.
+    /// Lazily computed and cached.
+    all_views_inlined: Option<bool>,
 }
 
 impl ViewBuffer {
@@ -37,6 +40,18 @@ impl ViewBuffer {
         self.views.is_empty()
     }
 
+    /// Computes and caches whether all views are inlined (length <= 12).
+    /// Call this after the buffer is fully built.
+    pub fn compute_all_views_inlined(&mut self) {
+        self.all_views_inlined = Some(self.views.iter().all(|&v| (v as u32) <= 12));
+    }
+
+    /// Returns the cached result of whether all views are inlined.
+    /// Returns `None` if `compute_all_views_inlined` has not been called.
+    pub fn all_views_inlined(&self) -> Option<bool> {
+        self.all_views_inlined
+    }
+
     pub fn append_block(&mut self, block: Buffer) -> u32 {
         let block_id = self.buffers.len() as u32;
         self.buffers.push(block);
diff --git a/parquet/src/arrow/decoder/dictionary_index.rs b/parquet/src/arrow/decoder/dictionary_index.rs
index ba16c5687560..2cda31db84d0 100644
--- a/parquet/src/arrow/decoder/dictionary_index.rs
+++ b/parquet/src/arrow/decoder/dictionary_index.rs
@@ -18,7 +18,7 @@
 use bytes::Bytes;
 
 use crate::encodings::rle::{RleDecodedBatch, RleDecoder};
-use crate::errors::Result;
+use crate::errors::{ParquetError, Result};
 
 /// Decoder for `Encoding::RLE_DICTIONARY` indices
 pub struct DictIndexDecoder {
@@ -105,12 +105,13 @@ impl DictIndexDecoder {
     ) -> Result<usize> {
         let to_read = len.min(self.max_remaining_values);
         let mut values_read = 0;
+        let mut error = None;
 
         // Flush any buffered indices from previous reads
         if self.index_offset < self.index_buf_len {
             let n = (self.index_buf_len - self.index_offset).min(to_read);
             let keys = &self.index_buf[self.index_offset..self.index_offset + n];
-            Self::gather_views(keys, dict_views, output, base_buffer_idx);
+            Self::gather_views(keys, dict_views, output, base_buffer_idx, &mut error);
             self.index_offset += n;
             self.max_remaining_values -= n;
             values_read += n;
@@ -121,24 +122,45 @@ impl DictIndexDecoder {
                 self.decoder
                     .get_batch_direct(to_read - values_read, |batch| match batch {
                         RleDecodedBatch::Rle { index, count } => {
-                            let view = dict_views[index as usize];
-                            let view = if base_buffer_idx == 0 || (view as u32) <= 12 {
-                                view
-                            } else {
-                                let mut bv = arrow_data::ByteView::from(view);
-                                bv.buffer_index += base_buffer_idx;
-                                bv.into()
-                            };
-                            output.extend(std::iter::repeat_n(view, count));
+                            match dict_views.get(index as usize) {
+                                Some(&view) => {
+                                    let view =
+                                        if base_buffer_idx == 0 || (view as u32) <= 12 {
+                                            view
+                                        } else {
+                                            let mut bv = arrow_data::ByteView::from(view);
+                                            bv.buffer_index += base_buffer_idx;
+                                            bv.into()
+                                        };
+                                    output.extend(std::iter::repeat_n(view, count));
+                                }
+                                None => {
+                                    if error.is_none() {
+                                        error = Some(general_err!(
+                                            "invalid key={} for dictionary",
+                                            index
+                                        ));
+                                    }
+                                }
+                            }
                         }
                         RleDecodedBatch::BitPacked(keys) => {
-                            Self::gather_views(keys, dict_views, output, base_buffer_idx);
+                            Self::gather_views(
+                                keys,
+                                dict_views,
+                                output,
+                                base_buffer_idx,
+                                &mut error,
+                            );
                         }
                     })?;
             self.max_remaining_values -= read;
             values_read += read;
         }
 
+        if let Some(e) = error {
+            return Err(e);
+        }
         Ok(values_read)
     }
 
@@ -147,24 +169,38 @@ impl DictIndexDecoder {
         dict_views: &[u128],
         output: &mut Vec<u128>,
         base_buffer_idx: u32,
+        error: &mut Option<ParquetError>,
     ) {
-        // Clamp index to valid range to prevent UB on corrupt data.
-        // This is branchless (cmp+csel on ARM) and avoids bounds checks in the hot loop.
-        let max_idx = dict_views.len() - 1;
         if base_buffer_idx == 0 {
-            output.extend(
-                keys.iter()
-                    .map(|k| unsafe { *dict_views.get_unchecked((*k as usize).min(max_idx)) }),
-            );
+            output.extend(keys.iter().map(|k| {
+                match dict_views.get(*k as usize) {
+                    Some(&view) => view,
+                    None => {
+                        if error.is_none() {
+                            *error = Some(general_err!("invalid key={} for dictionary", *k));
+                        }
+                        0
+                    }
+                }
+            }));
         } else {
             output.extend(keys.iter().map(|k| {
-                let view = unsafe { *dict_views.get_unchecked((*k as usize).min(max_idx)) };
-                if (view as u32) <= 12 {
-                    view
-                } else {
-                    let mut bv = arrow_data::ByteView::from(view);
-                    bv.buffer_index += base_buffer_idx;
-                    bv.into()
+                match dict_views.get(*k as usize) {
+                    Some(&view) => {
+                        if (view as u32) <= 12 {
+                            view
+                        } else {
+                            let mut bv = arrow_data::ByteView::from(view);
+                            bv.buffer_index += base_buffer_idx;
+                            bv.into()
+                        }
+                    }
+                    None => {
+                        if error.is_none() {
+                            *error = Some(general_err!("invalid key={} for dictionary", *k));
+                        }
+                        0
+                    }
                 }
             }));
         }

From 8ec819ed9289fcca401fa8541be802413c4e6bae Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Wed, 25 Mar 2026 19:13:05 +0100
Subject: [PATCH 13/16] Fmt

---
 parquet/src/arrow/decoder/dictionary_index.rs | 57 +++++++++----------
 1 file changed, 26 insertions(+), 31 deletions(-)

diff --git a/parquet/src/arrow/decoder/dictionary_index.rs b/parquet/src/arrow/decoder/dictionary_index.rs
index 2cda31db84d0..c17f1cc8dd07 100644
--- a/parquet/src/arrow/decoder/dictionary_index.rs
+++ b/parquet/src/arrow/decoder/dictionary_index.rs
@@ -124,14 +124,13 @@ impl DictIndexDecoder {
                         RleDecodedBatch::Rle { index, count } => {
                             match dict_views.get(index as usize) {
                                 Some(&view) => {
-                                    let view =
-                                        if base_buffer_idx == 0 || (view as u32) <= 12 {
-                                            view
-                                        } else {
-                                            let mut bv = arrow_data::ByteView::from(view);
-                                            bv.buffer_index += base_buffer_idx;
-                                            bv.into()
-                                        };
+                                    let view = if base_buffer_idx == 0 || (view as u32) <= 12 {
+                                        view
+                                    } else {
+                                        let mut bv = arrow_data::ByteView::from(view);
+                                        bv.buffer_index += base_buffer_idx;
+                                        bv.into()
+                                    };
                                     output.extend(std::iter::repeat_n(view, count));
                                 }
                                 None => {
@@ -172,35 +171,31 @@ impl DictIndexDecoder {
         error: &mut Option<ParquetError>,
     ) {
         if base_buffer_idx == 0 {
-            output.extend(keys.iter().map(|k| {
-                match dict_views.get(*k as usize) {
-                    Some(&view) => view,
-                    None => {
-                        if error.is_none() {
-                            *error = Some(general_err!("invalid key={} for dictionary", *k));
-                        }
-                        0
+            output.extend(keys.iter().map(|k| match dict_views.get(*k as usize) {
+                Some(&view) => view,
+                None => {
+                    if error.is_none() {
+                        *error = Some(general_err!("invalid key={} for dictionary", *k));
                     }
+                    0
                 }
             }));
         } else {
-            output.extend(keys.iter().map(|k| {
-                match dict_views.get(*k as usize) {
-                    Some(&view) => {
-                        if (view as u32) <= 12 {
-                            view
-                        } else {
-                            let mut bv = arrow_data::ByteView::from(view);
-                            bv.buffer_index += base_buffer_idx;
-                            bv.into()
-                        }
+            output.extend(keys.iter().map(|k| match dict_views.get(*k as usize) {
+                Some(&view) => {
+                    if (view as u32) <= 12 {
+                        view
+                    } else {
+                        let mut bv = arrow_data::ByteView::from(view);
+                        bv.buffer_index += base_buffer_idx;
+                        bv.into()
                     }
-                    None => {
-                        if error.is_none() {
-                            *error = Some(general_err!("invalid key={} for dictionary", *k));
-                        }
-                        0
+                }
+                None => {
+                    if error.is_none() {
+                        *error = Some(general_err!("invalid key={} for dictionary", *k));
                     }
+                    0
                 }
             }));
         }

From fca5b8ad51912be1dff5e64894745c0e700d9832 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Wed, 25 Mar 2026 19:13:48 +0100
Subject: [PATCH 14/16] Fmt

---
 parquet/src/arrow/buffer/dict_view_buffer.rs | 387 +++++++++++++++++++
 1 file changed, 387 insertions(+)
 create mode 100644 parquet/src/arrow/buffer/dict_view_buffer.rs

diff --git a/parquet/src/arrow/buffer/dict_view_buffer.rs b/parquet/src/arrow/buffer/dict_view_buffer.rs
new file mode 100644
index 000000000000..0810ae6d79af
--- /dev/null
+++ b/parquet/src/arrow/buffer/dict_view_buffer.rs
@@ -0,0 +1,387 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::arrow::buffer::view_buffer::ViewBuffer;
+use crate::arrow::record_reader::buffer::ValuesBuffer;
+use crate::errors::{ParquetError, Result};
+use arrow_array::{ArrayRef, StringViewArray};
+use arrow_buffer::Buffer;
+use arrow_data::ArrayDataBuilder;
+use arrow_schema::DataType as ArrowType;
+use std::sync::Arc;
+
+/// A buffer that can store variable-length byte view data either as dictionary-encoded
+/// (keys + dictionary) or as expanded views.
+///
+/// This is analogous to [`DictionaryBuffer`] but for view-based types (`Utf8View`, `BinaryView`).
+/// When dictionary encoding is preserved, the output is a `DictionaryArray<Int32, Utf8View>`.
+/// When dictionary encoding cannot be preserved (e.g., non-dictionary pages are encountered),
+/// the buffer falls back to expanded views.
+///
+/// [`DictionaryBuffer`]: crate::arrow::buffer::dictionary_buffer::DictionaryBuffer
+pub enum DictViewBuffer {
+    /// Dictionary-encoded: stores keys and a reference to the dictionary
+    Dict {
+        keys: Vec<i32>,
+        /// The dictionary as a StringViewArray (cheap Arc clone)
+        dict: ArrayRef,
+    },
+    /// Fallback: expanded views (same as ViewBuffer)
+    Values {
+        values: ViewBuffer,
+    },
+}
+
+impl Default for DictViewBuffer {
+    fn default() -> Self {
+        Self::Values {
+            values: ViewBuffer::default(),
+        }
+    }
+}
+
+impl std::fmt::Debug for DictViewBuffer {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Dict { keys, .. } => f
+                .debug_struct("DictViewBuffer::Dict")
+                .field("num_keys", &keys.len())
+                .finish(),
+            Self::Values { values } => f
+                .debug_struct("DictViewBuffer::Values")
+                .field("values", values)
+                .finish(),
+        }
+    }
+}
+
+impl DictViewBuffer {
+    #[allow(unused)]
+    pub fn len(&self) -> usize {
+        match self {
+            Self::Dict { keys, .. } => keys.len(),
+            Self::Values { values } => values.views.len(),
+        }
+    }
+
+    /// Returns a mutable reference to the keys vec if we can preserve the dictionary.
+    ///
+    /// Returns `None` if:
+    /// - The dictionary changed (different pointer) and keys are non-empty
+    /// - The buffer has already fallen back to Values mode with data
+    ///
+    /// When `None` is returned, the caller should use `spill_values()` instead.
+    pub fn as_keys(&mut self, dictionary: &ArrayRef) -> Option<&mut Vec<i32>> {
+        match self {
+            Self::Dict { keys, dict } => {
+                // Check if the dictionary is the same object
+                let dict_ptr = dict.as_ref() as *const _ as *const ();
+                let new_ptr = dictionary.as_ref() as *const _ as *const ();
+                if dict_ptr == new_ptr {
+                    Some(keys)
+                } else if keys.is_empty() {
+                    *dict = Arc::clone(dictionary);
+                    Some(keys)
+                } else {
+                    // Dictionary changed and we have existing keys - can't preserve
+                    None
+                }
+            }
+            Self::Values { values } if values.views.is_empty() => {
+                *self = Self::Dict {
+                    keys: Vec::new(),
+                    dict: Arc::clone(dictionary),
+                };
+                match self {
+                    Self::Dict { keys, .. } => Some(keys),
+                    _ => unreachable!(),
+                }
+            }
+            _ => None,
+        }
+    }
+
+    /// Spills dictionary-encoded data into expanded views.
+    ///
+    /// If already in `Values` mode, returns the existing `ViewBuffer`.
+    /// If in `Dict` mode, expands keys through the dictionary into views,
+    /// then switches to `Values` mode.
+    pub fn spill_values(&mut self) -> Result<&mut ViewBuffer> {
+        match self {
+            Self::Values { values } => Ok(values),
+            Self::Dict { keys, dict } => {
+                let mut spilled = ViewBuffer::default();
+
+                if !dict.is_empty() && !keys.is_empty() {
+                    // Expand the dictionary keys into views
+                    let dict_view = dict
+                        .as_any()
+                        .downcast_ref::<StringViewArray>()
+                        .ok_or_else(|| {
+                            general_err!("DictViewBuffer dictionary is not StringViewArray")
+                        })?;
+
+                    // Copy dictionary buffers to the output
+                    for buf in dict_view.data_buffers() {
+                        spilled.buffers.push(buf.clone());
+                    }
+
+                    let dict_views = dict_view.views();
+                    spilled.views.reserve(keys.len());
+
+                    for &key in keys.iter() {
+                        if key < 0 || (key as usize) >= dict_views.len() {
+                            return Err(general_err!(
+                                "dictionary key {} beyond bounds of dictionary: 0..{}",
+                                key,
+                                dict_views.len()
+                            ));
+                        }
+                        let view = dict_views[key as usize];
+                        // Adjust buffer indices since dict buffers are the first buffers
+                        // in the output (base_buffer_idx = 0)
+                        spilled.views.push(view);
+                    }
+                } else if !keys.is_empty() {
+                    // Empty dictionary but non-empty keys - zero-length views
+                    spilled.views.resize(keys.len(), 0u128);
+                }
+
+                *self = Self::Values { values: spilled };
+                match self {
+                    Self::Values { values } => Ok(values),
+                    _ => unreachable!(),
+                }
+            }
+        }
+    }
+
+    /// Converts this buffer into an [`ArrayRef`].
+    ///
+    /// When in `Dict` mode, produces a `DictionaryArray<Int32, Utf8View>`.
+    /// When in `Values` mode, packs the values into a fresh `DictionaryArray`.
+    pub fn into_array(self, null_buffer: Option<Buffer>, data_type: &ArrowType) -> Result<ArrayRef> {
+        assert!(
+            matches!(data_type, ArrowType::Dictionary(_, _)),
+            "DictViewBuffer requires Dictionary data type, got {data_type}"
+        );
+
+        match self {
+            Self::Dict { keys, dict } => {
+                // Validate keys
+                if !dict.is_empty() {
+                    let max = dict.len();
+                    if !keys
+                        .iter()
+                        .copied()
+                        .fold(true, |a, x| a && x >= 0 && (x as usize) < max)
+                    {
+                        return Err(general_err!(
+                            "dictionary key beyond bounds of dictionary: 0..{}",
+                            max
+                        ));
+                    }
+                }
+
+                let builder = ArrayDataBuilder::new(data_type.clone())
+                    .len(keys.len())
+                    .add_buffer(Buffer::from_vec(keys))
+                    .add_child_data(dict.to_data())
+                    .null_bit_buffer(null_buffer);
+
+                let data = match cfg!(debug_assertions) {
+                    true => builder.build().unwrap(),
+                    false => unsafe { builder.build_unchecked() },
+                };
+
+                Ok(arrow_array::make_array(data))
+            }
+            Self::Values { values } => {
+                // Need to create a dictionary from expanded values
+                let ArrowType::Dictionary(_, value_type) = data_type else {
+                    unreachable!()
+                };
+                let values_array = values.into_array(null_buffer, value_type);
+                pack_string_view_values(&values_array, data_type)
+            }
+        }
+    }
+}
+
+/// Pack a `StringViewArray` into a `DictionaryArray<Int32, StringViewArray>`.
+///
+/// This creates an identity-mapping dictionary where each row maps to its
+/// own dictionary entry. This is a fallback for when dictionary encoding
+/// couldn't be preserved, and is suboptimal but correct.
+fn pack_string_view_values(array: &ArrayRef, dict_type: &ArrowType) -> Result<ArrayRef> {
+    let len = array.len();
+
+    // Create identity keys [0, 1, 2, ..., len-1]
+    let keys: Vec<i32> = (0..len as i32).collect();
+
+    let builder = ArrayDataBuilder::new(dict_type.clone())
+        .len(len)
+        .add_buffer(Buffer::from_vec(keys))
+        .add_child_data(array.to_data())
+        .null_bit_buffer(array.nulls().map(|n| n.buffer().clone()));
+
+    let data = match cfg!(debug_assertions) {
+        true => builder.build().unwrap(),
+        false => unsafe { builder.build_unchecked() },
+    };
+
+    Ok(arrow_array::make_array(data))
+}
+
+impl ValuesBuffer for DictViewBuffer {
+    fn pad_nulls(
+        &mut self,
+        read_offset: usize,
+        values_read: usize,
+        levels_read: usize,
+        valid_mask: &[u8],
+    ) {
+        match self {
+            Self::Dict { keys, .. } => {
+                keys.resize(read_offset + levels_read, 0i32);
+                keys.pad_nulls(read_offset, values_read, levels_read, valid_mask);
+            }
+            Self::Values { values } => {
+                values.pad_nulls(read_offset, values_read, levels_read, valid_mask);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow::compute::cast;
+    use arrow_array::{Array, StringViewArray};
+    use arrow_buffer::Buffer;
+
+    fn utf8view_dictionary() -> ArrowType {
+        ArrowType::Dictionary(Box::new(ArrowType::Int32), Box::new(ArrowType::Utf8View))
+    }
+
+    #[test]
+    fn test_dict_view_buffer_dict_mode() {
+        let data_type = utf8view_dictionary();
+
+        // Create a dictionary
+        let dict: ArrayRef = Arc::new(StringViewArray::from(vec!["hello", "world", ""]));
+
+        let mut buffer = DictViewBuffer::default();
+
+        // Write keys preserving dictionary
+        let keys = buffer.as_keys(&dict).unwrap();
+        keys.extend_from_slice(&[1, 0, 2, 0]);
+
+        assert!(matches!(buffer, DictViewBuffer::Dict { .. }));
+
+        let array = buffer.into_array(None, &data_type).unwrap();
+        assert_eq!(array.data_type(), &data_type);
+
+        let strings = cast(&array, &ArrowType::Utf8View).unwrap();
+        let strings = strings
+            .as_any()
+            .downcast_ref::<StringViewArray>()
+            .unwrap();
+        assert_eq!(
+            strings.iter().collect::<Vec<_>>(),
+            vec![Some("world"), Some("hello"), Some(""), Some("hello")]
+        );
+    }
+
+    #[test]
+    fn test_dict_view_buffer_with_nulls() {
+        let data_type = utf8view_dictionary();
+
+        let dict: ArrayRef = Arc::new(StringViewArray::from(vec!["a", "b"]));
+
+        let mut buffer = DictViewBuffer::default();
+        let keys = buffer.as_keys(&dict).unwrap();
+        keys.extend_from_slice(&[0, 1]);
+
+        let valid = [false, true, true, false];
+        let valid_buffer = Buffer::from_iter(valid.iter().copied());
+        buffer.pad_nulls(0, 2, valid.len(), valid_buffer.as_slice());
+
+        let array = buffer
+            .into_array(Some(valid_buffer), &data_type)
+            .unwrap();
+        assert_eq!(array.len(), 4);
+        assert_eq!(array.null_count(), 2);
+
+        let strings = cast(&array, &ArrowType::Utf8View).unwrap();
+        let strings = strings
+            .as_any()
+            .downcast_ref::<StringViewArray>()
+            .unwrap();
+        assert_eq!(
+            strings.iter().collect::<Vec<_>>(),
+            vec![None, Some("a"), Some("b"), None]
+        );
+    }
+
+    #[test]
+    fn test_dict_view_buffer_spill() {
+        let data_type = utf8view_dictionary();
+
+        let dict: ArrayRef = Arc::new(StringViewArray::from(vec!["hello", "world"]));
+
+        let mut buffer = DictViewBuffer::default();
+        let keys = buffer.as_keys(&dict).unwrap();
+        keys.extend_from_slice(&[0, 1]);
+
+        // Spill to values mode
+        let _values = buffer.spill_values().unwrap();
+
+        assert!(matches!(buffer, DictViewBuffer::Values { .. }));
+
+        let array = buffer.into_array(None, &data_type).unwrap();
+        assert_eq!(array.data_type(), &data_type);
+
+        let strings = cast(&array, &ArrowType::Utf8View).unwrap();
+        let strings = strings
+            .as_any()
+            .downcast_ref::<StringViewArray>()
+            .unwrap();
+        assert_eq!(
+            strings.iter().collect::<Vec<_>>(),
+            vec![Some("hello"), Some("world")]
+        );
+    }
+
+    #[test]
+    fn test_dict_view_buffer_validates_keys() {
+        let data_type = utf8view_dictionary();
+
+        let dict: ArrayRef = Arc::new(StringViewArray::from(vec!["a"]));
+
+        let mut buffer = DictViewBuffer::default();
+        let keys = buffer.as_keys(&dict).unwrap();
+        keys.extend_from_slice(&[0, 2, 0]); // key 2 is out of bounds
+
+        let err = buffer.into_array(None, &data_type).unwrap_err().to_string();
+        assert!(
+            err.contains("dictionary key beyond bounds"),
+            "{}",
+            err
+        );
+    }
+}

From 81da338022f4694dd899055bfbd3a319a7e64d6b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Wed, 25 Mar 2026 19:25:34 +0100
Subject: [PATCH 15/16] Remove all_views_inlined caching from ViewBuffer

The per-view inline check in gather_views already handles inlined views
correctly, making the precomputed cache unnecessary complexity.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 .../src/arrow/array_reader/byte_view_array.rs | 47 +++++++------------
 parquet/src/arrow/buffer/view_buffer.rs       | 15 ------
 2 files changed, 17 insertions(+), 45 deletions(-)

diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index cb27b85bfd11..7c1cf145100f 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -169,7 +169,6 @@ impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
             self.validate_utf8,
         );
         decoder.read(&mut buffer, usize::MAX)?;
-        buffer.compute_all_views_inlined();
         self.dict = Some(buffer);
         Ok(())
     }
@@ -475,40 +474,28 @@ impl ByteViewArrayDecoderDictionary {
             return Ok(0);
         }
 
-        // Check if all dictionary views are inlined (len <= 12).
-        // If so, we can skip buffer management entirely since inline views
-        // don't reference any buffers.
-        // This value is precomputed when the dictionary is set.
-        let all_views_inlined = dict.all_views_inlined().unwrap_or(false);
-
-        if !all_views_inlined {
-            // Check if the last few buffer of `output`` are the same as the `dict` buffer
-            // This is to avoid creating a new buffers if the same dictionary is used for multiple `read`
-            let need_to_create_new_buffer = {
-                if output.buffers.len() >= dict.buffers.len() {
-                    let offset = output.buffers.len() - dict.buffers.len();
-                    output.buffers[offset..]
-                        .iter()
-                        .zip(dict.buffers.iter())
-                        .any(|(a, b)| !a.ptr_eq(b))
-                } else {
-                    true
-                }
-            };
+        // Check if the last few buffer of `output`` are the same as the `dict` buffer
+        // This is to avoid creating a new buffers if the same dictionary is used for multiple `read`
+        let need_to_create_new_buffer = {
+            if output.buffers.len() >= dict.buffers.len() {
+                let offset = output.buffers.len() - dict.buffers.len();
+                output.buffers[offset..]
+                    .iter()
+                    .zip(dict.buffers.iter())
+                    .any(|(a, b)| !a.ptr_eq(b))
+            } else {
+                true
+            }
+        };
 
-            if need_to_create_new_buffer {
-                for b in dict.buffers.iter() {
-                    output.buffers.push(b.clone());
-                }
+        if need_to_create_new_buffer {
+            for b in dict.buffers.iter() {
+                output.buffers.push(b.clone());
             }
         }
 
         // Calculate the offset of the dictionary buffers in the output buffers
-        let base_buffer_idx = if all_views_inlined {
-            0
-        } else {
-            output.buffers.len() as u32 - dict.buffers.len() as u32
-        };
+        let base_buffer_idx = output.buffers.len() as u32 - dict.buffers.len() as u32;
 
         // Pre-reserve output capacity to avoid per-chunk reallocation in extend
         output.views.reserve(len);
diff --git a/parquet/src/arrow/buffer/view_buffer.rs b/parquet/src/arrow/buffer/view_buffer.rs
index 1e73b86ce0b3..a93674663f7b 100644
--- a/parquet/src/arrow/buffer/view_buffer.rs
+++ b/parquet/src/arrow/buffer/view_buffer.rs
@@ -30,9 +30,6 @@ use std::sync::Arc;
 pub struct ViewBuffer {
     pub views: Vec<u128>,
     pub buffers: Vec<Buffer>,
-    /// Whether all views are inlined (length <= 12), meaning no buffer references are needed.
-    /// Lazily computed and cached.
-    all_views_inlined: Option<bool>,
 }
 
 impl ViewBuffer {
@@ -40,18 +37,6 @@ impl ViewBuffer {
         self.views.is_empty()
     }
 
-    /// Computes and caches whether all views are inlined (length <= 12).
-    /// Call this after the buffer is fully built.
-    pub fn compute_all_views_inlined(&mut self) {
-        self.all_views_inlined = Some(self.views.iter().all(|&v| (v as u32) <= 12));
-    }
-
-    /// Returns the cached result of whether all views are inlined.
-    /// Returns `None` if `compute_all_views_inlined` has not been called.
-    pub fn all_views_inlined(&self) -> Option<bool> {
-        self.all_views_inlined
-    }
-
     pub fn append_block(&mut self, block: Buffer) -> u32 {
         let block_id = self.buffers.len() as u32;
         self.buffers.push(block);

From fb7cce76f4a4a7dd498e2dd2fcbacde75e88c661 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dani=C3=ABl=20Heres?= <danielheres@gmail.com>
Date: Wed, 25 Mar 2026 19:35:56 +0100
Subject: [PATCH 16/16] Remove branchless clamping in RLE dict decode, use
 bounds-checked indexing

The clamping with get_unchecked + min(max_idx) silently returned wrong
values for corrupt data. Normal indexing will panic on invalid keys,
which is the correct behavior.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 parquet/src/encodings/rle.rs | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index 5f11cc51fcb8..ff79402969cf 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -516,10 +516,6 @@ impl RleDecoder {
         if dict.is_empty() {
             return Ok(0);
         }
-        // Clamp index to valid range to prevent UB on corrupt data.
-        // This is branchless (cmp+csel on ARM) and avoids bounds checks in the hot loop.
-        let max_idx = dict.len() - 1;
-
         let mut values_read = 0;
         while values_read < max_values {
             let index_buf = self.index_buf.get_or_insert_with(|| Box::new([0; 1024]));
@@ -559,8 +555,8 @@ impl RleDecoder {
                     buffer[values_read..values_read + num_values]
                         .iter_mut()
                         .zip(index_buf[..num_values].iter())
-                        .for_each(|(b, i)| unsafe {
-                            b.clone_from(dict.get_unchecked((*i as usize).min(max_idx)));
+                        .for_each(|(b, i)| {
+                            b.clone_from(&dict[*i as usize]);
                         });
                     self.bit_packed_left -= num_values as u32;
                     values_read += num_values;