Repository: apache/arrow-rs
Pull request #6019 (2024-07-08 13:44:59)

Update zstd-sys requirement from >=2.0.0, <2.0.12 to >=2.0.0, <2.0.13
Updates the requirements on [zstd-sys](https://github.com/gyscos/zstd-rs) to permit the latest version.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/gyscos/zstd-rs/commit/46d8f58001e8536fc86e02b4d0c9ec45469933aa"><code>46d8f58</code></a> Bump zstd-sys to 2.0.12</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/6ad8559b4361a0b6fcd0f3c3a742493ce14c73ce"><code>6ad8559</code></a> Fix merge issues</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/66882071ad023e3a3dd4538bedee1f18932dbb6b"><code>6688207</code></a> Add the ability to provide a safe context for encoders to use</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/a3738d680542e34b5529207ee30491ed7b69ed71"><code>a3738d6</code></a> remove hardcoded BufRead trait bounds (<a href="https://redirect.github.com/gyscos/zstd-rs/issues/288">#288</a>)</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/cc5039d734fbbc430323987fd90c7417db111492"><code>cc5039d</code></a> chore: fix some comments (<a href="https://redirect.github.com/gyscos/zstd-rs/issues/287">#287</a>)</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/8e7aa79ee2eeeb36567dc56158371bc68b0e3796"><code>8e7aa79</code></a> Bump zstd-sys to 2.0.11</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/173c956dfec1e85dda2b8bdbc91272a3b5baacb4"><code>173c956</code></a> Support thread pools (<a href="https://redirect.github.com/gyscos/zstd-rs/issues/283">#283</a>)</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/2b63859f04732add93f740c07ef29e4b3e50d1af"><code>2b63859</code></a> Make ref_prefix available (<a href="https://redirect.github.com/gyscos/zstd-rs/issues/273">#273</a>)</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/fb6b490b7cf991216e815ca3f63b20ae037c9b4b"><code>fb6b490</code></a> wasm32-unknown-unknown: Fix undefined malloc/free/calloc symbols (<a href="https://redirect.github.com/gyscos/zstd-rs/issues/275">#275</a>)</li>
<li><a href="https://github.com/gyscos/zstd-rs/commit/a2125e3ebcc142928f723e7099fed3e24738f5b0"><code>a2125e3</code></a> Fix wasm32-unknown-unknown support (<a href="https://redirect.github.com/gyscos/zstd-rs/issues/274">#274</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/gyscos/zstd-rs/compare/zstd-sys-2.0.7...zstd-sys-2.0.12">compare view</a></li>
</ul>
</details>
<br />


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index c07821959..d0a78ca27 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -411,7 +411,7 @@ impl IpcDataGenerator {
         write_options: &IpcWriteOptions,
     ) -> Result<(Vec<EncodedData>, EncodedData), ArrowError> {
         let schema = batch.schema();
-        let mut encoded_dictionaries = Vec::with_capacity(schema.all_fields().len());
+        let mut encoded_dictionaries = Vec::with_capacity(schema.flattened_fields().len());
 
         let mut dict_id = dictionary_tracker.dict_ids.clone().into_iter();
 
diff --git a/arrow-json/src/reader/mod.rs b/arrow-json/src/reader/mod.rs
index 0a50cfac6..3e1c5d2fc 100644
--- a/arrow-json/src/reader/mod.rs
+++ b/arrow-json/src/reader/mod.rs
@@ -289,7 +289,7 @@ impl ReaderBuilder {
 
         let decoder = make_decoder(data_type, self.coerce_primitive, self.strict_mode, nullable)?;
 
-        let num_fields = self.schema.all_fields().len();
+        let num_fields = self.schema.flattened_fields().len();
 
         Ok(Decoder {
             decoder,
diff --git a/arrow-schema/src/schema.rs b/arrow-schema/src/schema.rs
index 3bb076aa5..9a9ef45d8 100644
--- a/arrow-schema/src/schema.rs
+++ b/arrow-schema/src/schema.rs
@@ -324,11 +324,53 @@ impl Schema {
     }
 
     /// Returns a vector with references to all fields (including nested fields)
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use std::sync::Arc;
+    /// use arrow_schema::{DataType, Field, Fields, Schema};
+    ///
+    /// let f1 = Arc::new(Field::new("a", DataType::Boolean, false));
+    ///
+    /// let f2_inner = Arc::new(Field::new("b_inner", DataType::Int8, false));
+    /// let f2 = Arc::new(Field::new("b", DataType::List(f2_inner.clone()), false));
+    ///
+    /// let f3_inner1 = Arc::new(Field::new("c_inner1", DataType::Int8, false));
+    /// let f3_inner2 = Arc::new(Field::new("c_inner2", DataType::Int8, false));
+    /// let f3 = Arc::new(Field::new(
+    ///     "c",
+    ///     DataType::Struct(vec![f3_inner1.clone(), f3_inner2.clone()].into()),
+    ///     false
+    /// ));
+    ///
+    /// let mut schema = Schema::new(vec![
+    ///   f1.clone(), f2.clone(), f3.clone()
+    /// ]);
+    /// assert_eq!(
+    ///     schema.flattened_fields(),
+    ///     vec![
+    ///         f1.as_ref(),
+    ///         f2.as_ref(),
+    ///         f2_inner.as_ref(),
+    ///         f3.as_ref(),
+    ///         f3_inner1.as_ref(),
+    ///         f3_inner2.as_ref()
+    ///    ]
+    /// );
+    /// ```
     #[inline]
-    pub fn all_fields(&self) -> Vec<&Field> {
+    pub fn flattened_fields(&self) -> Vec<&Field> {
         self.fields.iter().flat_map(|f| f.fields()).collect()
     }
 
+    /// Returns a vector with references to all fields (including nested fields)
+    #[deprecated(since = "52.2.0", note = "Use `flattened_fields` instead")]
+    #[inline]
+    pub fn all_fields(&self) -> Vec<&Field> {
+        self.flattened_fields()
+    }
+
     /// Returns an immutable reference of a specific [`Field`] instance selected using an
     /// offset within the internal `fields` vector.
     ///
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 7df978016..2fb808e1e 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -86,10 +86,10 @@ object_store = { version = "0.10.0", default-features = false, features = ["azur
 # TODO: temporary to fix parquet wasm build
 # upstream issue: https://github.com/gyscos/zstd-rs/issues/269
 [target.'cfg(target_family = "wasm")'.dependencies]
-zstd-sys = { version = ">=2.0.0, <2.0.12", optional = true, default-features = false }
+zstd-sys = { version = ">=2.0.0, <2.0.13", optional = true, default-features = false }
 
 [target.'cfg(target_family = "wasm")'.dev-dependencies]
-zstd-sys = { version = ">=2.0.0, <2.0.12", default-features = false }
+zstd-sys = { version = ">=2.0.0, <2.0.13", default-features = false }
 
 [package.metadata.docs.rs]
 all-features = true
diff --git a/parquet/src/arrow/array_reader/builder.rs b/parquet/src/arrow/array_reader/builder.rs
index 958594c93..945f62526 100644
--- a/parquet/src/arrow/array_reader/builder.rs
+++ b/parquet/src/arrow/array_reader/builder.rs
@@ -19,7 +19,7 @@ use std::sync::Arc;
 
 use arrow_schema::{DataType, Fields, SchemaBuilder};
 
-use crate::arrow::array_reader::byte_array::make_byte_view_array_reader;
+use crate::arrow::array_reader::byte_view_array::make_byte_view_array_reader;
 use crate::arrow::array_reader::empty_array::make_empty_array_reader;
 use crate::arrow::array_reader::fixed_len_byte_array::make_fixed_len_byte_array_reader;
 use crate::arrow::array_reader::{
diff --git a/parquet/src/arrow/array_reader/byte_array.rs b/parquet/src/arrow/array_reader/byte_array.rs
index d0aa6f7b1..80c448edc 100644
--- a/parquet/src/arrow/array_reader/byte_array.rs
+++ b/parquet/src/arrow/array_reader/byte_array.rs
@@ -74,36 +74,6 @@ pub fn make_byte_array_reader(
     }
 }
 
-/// Returns an [`ArrayReader`] that decodes the provided byte array column to view types.
-pub fn make_byte_view_array_reader(
-    pages: Box<dyn PageIterator>,
-    column_desc: ColumnDescPtr,
-    arrow_type: Option<ArrowType>,
-) -> Result<Box<dyn ArrayReader>> {
-    // Check if Arrow type is specified, else create it from Parquet type
-    let data_type = match arrow_type {
-        Some(t) => t,
-        None => match parquet_to_arrow_field(column_desc.as_ref())?.data_type() {
-            ArrowType::Utf8 | ArrowType::Utf8View => ArrowType::Utf8View,
-            _ => ArrowType::BinaryView,
-        },
-    };
-
-    match data_type {
-        ArrowType::BinaryView | ArrowType::Utf8View => {
-            let reader = GenericRecordReader::new(column_desc);
-            Ok(Box::new(ByteArrayReader::<i32>::new(
-                pages, data_type, reader,
-            )))
-        }
-
-        _ => Err(general_err!(
-            "invalid data type for byte array reader read to view type - {}",
-            data_type
-        )),
-    }
-}
-
 /// An [`ArrayReader`] for variable length byte arrays
 struct ByteArrayReader<I: OffsetSizeTrait> {
     data_type: ArrowType,
@@ -472,6 +442,23 @@ impl ByteArrayDecoderDeltaLength {
         let mut lengths = vec![0; values];
         len_decoder.get(&mut lengths)?;
 
+        let mut total_bytes = 0;
+
+        for l in lengths.iter() {
+            if *l < 0 {
+                return Err(ParquetError::General(
+                    "negative delta length byte array length".to_string(),
+                ));
+            }
+            total_bytes += *l as usize;
+        }
+
+        if total_bytes + len_decoder.get_offset() > data.len() {
+            return Err(ParquetError::General(
+                "Insufficient delta length byte array bytes".to_string(),
+            ));
+        }
+
         Ok(Self {
             lengths,
             data,
@@ -496,23 +483,17 @@ impl ByteArrayDecoderDeltaLength {
         let total_bytes: usize = src_lengths.iter().map(|x| *x as usize).sum();
         output.values.reserve(total_bytes);
 
-        if self.data_offset + total_bytes > self.data.len() {
-            return Err(ParquetError::EOF(
-                "Insufficient delta length byte array bytes".to_string(),
-            ));
-        }
-
-        let mut start_offset = self.data_offset;
+        let mut current_offset = self.data_offset;
         for length in src_lengths {
-            let end_offset = start_offset + *length as usize;
+            let end_offset = current_offset + *length as usize;
             output.try_push(
-                &self.data.as_ref()[start_offset..end_offset],
+                &self.data.as_ref()[current_offset..end_offset],
                 self.validate_utf8,
             )?;
-            start_offset = end_offset;
+            current_offset = end_offset;
         }
 
-        self.data_offset = start_offset;
+        self.data_offset = current_offset;
         self.length_offset += to_read;
 
         if self.validate_utf8 {
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 7f0a0dd2a..dc4ce3f9c 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -17,22 +17,23 @@
 
 use crate::arrow::array_reader::{read_records, skip_records, ArrayReader};
 use crate::arrow::buffer::view_buffer::ViewBuffer;
-use crate::arrow::decoder::DictIndexDecoder;
+use crate::arrow::decoder::{DeltaByteArrayDecoder, DictIndexDecoder};
 use crate::arrow::record_reader::GenericRecordReader;
 use crate::arrow::schema::parquet_to_arrow_field;
 use crate::basic::{ConvertedType, Encoding};
 use crate::column::page::PageIterator;
 use crate::column::reader::decoder::ColumnValueDecoder;
+use crate::data_type::Int32Type;
+use crate::encodings::decoding::{Decoder, DeltaBitPackDecoder};
 use crate::errors::{ParquetError, Result};
 use crate::schema::types::ColumnDescPtr;
-use arrow_array::ArrayRef;
+use arrow_array::{builder::make_view, ArrayRef};
 use arrow_data::ByteView;
 use arrow_schema::DataType as ArrowType;
 use bytes::Bytes;
 use std::any::Any;
 
 /// Returns an [`ArrayReader`] that decodes the provided byte array column to view types.
-#[allow(unused)]
 pub fn make_byte_view_array_reader(
     pages: Box<dyn PageIterator>,
     column_desc: ColumnDescPtr,
@@ -61,7 +62,6 @@ pub fn make_byte_view_array_reader(
 }
 
 /// An [`ArrayReader`] for variable length byte arrays
-#[allow(unused)]
 struct ByteViewArrayReader {
     data_type: ArrowType,
     pages: Box<dyn PageIterator>,
@@ -213,6 +213,8 @@ impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
 pub enum ByteViewArrayDecoder {
     Plain(ByteViewArrayDecoderPlain),
     Dictionary(ByteViewArrayDecoderDictionary),
+    DeltaLength(ByteViewArrayDecoderDeltaLength),
+    DeltaByteArray(ByteViewArrayDecoderDelta),
 }
 
 impl ByteViewArrayDecoder {
@@ -235,9 +237,12 @@ impl ByteViewArrayDecoder {
                     data, num_levels, num_values,
                 ))
             }
-            Encoding::DELTA_LENGTH_BYTE_ARRAY | Encoding::DELTA_BYTE_ARRAY => {
-                unimplemented!("stay tuned!")
-            }
+            Encoding::DELTA_LENGTH_BYTE_ARRAY => ByteViewArrayDecoder::DeltaLength(
+                ByteViewArrayDecoderDeltaLength::new(data, validate_utf8)?,
+            ),
+            Encoding::DELTA_BYTE_ARRAY => ByteViewArrayDecoder::DeltaByteArray(
+                ByteViewArrayDecoderDelta::new(data, validate_utf8)?,
+            ),
             _ => {
                 return Err(general_err!(
                     "unsupported encoding for byte array: {}",
@@ -263,6 +268,8 @@ impl ByteViewArrayDecoder {
                     .ok_or_else(|| general_err!("dictionary required for dictionary encoding"))?;
                 d.read(out, dict, len)
             }
+            ByteViewArrayDecoder::DeltaLength(d) => d.read(out, len),
+            ByteViewArrayDecoder::DeltaByteArray(d) => d.read(out, len),
         }
     }
 
@@ -275,6 +282,8 @@ impl ByteViewArrayDecoder {
                     .ok_or_else(|| general_err!("dictionary required for dictionary encoding"))?;
                 d.skip(dict, len)
             }
+            ByteViewArrayDecoder::DeltaLength(d) => d.skip(len),
+            ByteViewArrayDecoder::DeltaByteArray(d) => d.skip(len),
         }
     }
 }
@@ -314,6 +323,8 @@ impl ByteViewArrayDecoderPlain {
         let buf = self.buf.as_ref();
         let mut read = 0;
         output.views.reserve(to_read);
+
+        let mut utf8_validation_begin = self.offset;
         while self.offset < self.buf.len() && read != to_read {
             if self.offset + 4 > self.buf.len() {
                 return Err(ParquetError::EOF("eof decoding byte array".into()));
@@ -332,7 +343,38 @@ impl ByteViewArrayDecoderPlain {
             }
 
             if self.validate_utf8 {
-                check_valid_utf8(unsafe { buf.get_unchecked(start_offset..end_offset) })?;
+                // It seems you are trying to understand what's going on here, take a breath and be patient.
+                // Utf-8 validation is a non-trivial task, here are some background facts:
+                // (1) Validating one 2048-byte string is much faster than validating 128 of 16-byte string.
+                //     As shown in https://github.com/apache/arrow-rs/pull/6009#issuecomment-2211174229
+                //     Potentially because the SIMD operations favor longer strings.
+                // (2) Practical strings are short, 99% of strings are smaller than 100 bytes, as shown in paper:
+                //     https://www.vldb.org/pvldb/vol17/p148-zeng.pdf, Figure 5f.
+                // (3) Parquet plain encoding makes utf-8 validation harder,
+                //     because it stores the length of each string right before the string.
+                //     This means naive utf-8 validation will be slow, because the validation need to skip the length bytes.
+                //     I.e., the validation cannot validate the buffer in one pass, but instead, validate strings chunk by chunk.
+                //
+                // Given the above observations, the goal is to do batch validation as much as possible.
+                // The key idea is that if the length is smaller than 128 (99% of the case), then the length bytes are valid utf-8, as reasoned below:
+                // If the length is smaller than 128, its 4-byte encoding are [0, 0, 0, len].
+                // Each of the byte is a valid ASCII character, so they are valid utf-8.
+                // Since they are all smaller than 128, the won't break a utf-8 code point (won't mess with later bytes).
+                //
+                // The implementation keeps a water mark `utf8_validation_begin` to track the beginning of the buffer that is not validated.
+                // If the length is smaller than 128, then we continue to next string.
+                // If the length is larger than 128, then we validate the buffer before the length bytes, and move the water mark to the beginning of next string.
+                if len < 128 {
+                    // fast path, move to next string.
+                    // the len bytes are valid utf8.
+                } else {
+                    // unfortunately, the len bytes may not be valid utf8, we need to wrap up and validate everything before it.
+                    check_valid_utf8(unsafe {
+                        buf.get_unchecked(utf8_validation_begin..self.offset)
+                    })?;
+                    // move the cursor to skip the len bytes.
+                    utf8_validation_begin = start_offset;
+                }
             }
 
             unsafe {
@@ -342,6 +384,11 @@ impl ByteViewArrayDecoderPlain {
             read += 1;
         }
 
+        // validate the last part of the buffer
+        if self.validate_utf8 {
+            check_valid_utf8(unsafe { buf.get_unchecked(utf8_validation_begin..self.offset) })?;
+        }
+
         self.max_remaining_values -= to_read;
         Ok(to_read)
     }
@@ -449,6 +496,181 @@ impl ByteViewArrayDecoderDictionary {
     }
 }
 
+/// Decoder from [`Encoding::DELTA_LENGTH_BYTE_ARRAY`] data to [`ViewBuffer`]
+pub struct ByteViewArrayDecoderDeltaLength {
+    lengths: Vec<i32>,
+    data: Bytes,
+    length_offset: usize,
+    data_offset: usize,
+    validate_utf8: bool,
+}
+
+impl ByteViewArrayDecoderDeltaLength {
+    fn new(data: Bytes, validate_utf8: bool) -> Result<Self> {
+        let mut len_decoder = DeltaBitPackDecoder::<Int32Type>::new();
+        len_decoder.set_data(data.clone(), 0)?;
+        let values = len_decoder.values_left();
+
+        let mut lengths = vec![0; values];
+        len_decoder.get(&mut lengths)?;
+
+        let mut total_bytes = 0;
+
+        for l in lengths.iter() {
+            if *l < 0 {
+                return Err(ParquetError::General(
+                    "negative delta length byte array length".to_string(),
+                ));
+            }
+            total_bytes += *l as usize;
+        }
+
+        if total_bytes + len_decoder.get_offset() > data.len() {
+            return Err(ParquetError::General(
+                "Insufficient delta length byte array bytes".to_string(),
+            ));
+        }
+
+        Ok(Self {
+            lengths,
+            data,
+            validate_utf8,
+            length_offset: 0,
+            data_offset: len_decoder.get_offset(),
+        })
+    }
+
+    fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
+        let to_read = len.min(self.lengths.len() - self.length_offset);
+        output.views.reserve(to_read);
+
+        let src_lengths = &self.lengths[self.length_offset..self.length_offset + to_read];
+
+        let block_id = output.append_block(self.data.clone().into());
+
+        let mut current_offset = self.data_offset;
+        let initial_offset = current_offset;
+        for length in src_lengths {
+            // # Safety
+            // The length is from the delta length decoder, so it is valid
+            // The start_offset is calculated from the lengths, so it is valid
+            // `start_offset + length` is guaranteed to be within the bounds of `data`, as checked in `new`
+            unsafe { output.append_view_unchecked(block_id, current_offset as u32, *length as u32) }
+
+            current_offset += *length as usize;
+        }
+
+        // Delta length encoding has continuous strings, we can validate utf8 in one go
+        if self.validate_utf8 {
+            check_valid_utf8(&self.data[initial_offset..current_offset])?;
+        }
+
+        self.data_offset = current_offset;
+        self.length_offset += to_read;
+
+        Ok(to_read)
+    }
+
+    fn skip(&mut self, to_skip: usize) -> Result<usize> {
+        let remain_values = self.lengths.len() - self.length_offset;
+        let to_skip = remain_values.min(to_skip);
+
+        let src_lengths = &self.lengths[self.length_offset..self.length_offset + to_skip];
+        let total_bytes: usize = src_lengths.iter().map(|x| *x as usize).sum();
+
+        self.data_offset += total_bytes;
+        self.length_offset += to_skip;
+        Ok(to_skip)
+    }
+}
+
+/// Decoder from [`Encoding::DELTA_BYTE_ARRAY`] to [`ViewBuffer`]
+pub struct ByteViewArrayDecoderDelta {
+    decoder: DeltaByteArrayDecoder,
+    validate_utf8: bool,
+}
+
+impl ByteViewArrayDecoderDelta {
+    fn new(data: Bytes, validate_utf8: bool) -> Result<Self> {
+        Ok(Self {
+            decoder: DeltaByteArrayDecoder::new(data)?,
+            validate_utf8,
+        })
+    }
+
+    // Unlike other encodings, we need to copy the data.
+    //
+    //  DeltaByteArray data is stored using shared prefixes/suffixes,
+    // which results in potentially non-contiguous
+    // strings, while Arrow encodings require contiguous strings
+    //
+    // <https://parquet.apache.org/docs/file-format/data-pages/encodings/#delta-strings-delta_byte_array--7>
+
+    fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
+        output.views.reserve(len.min(self.decoder.remaining()));
+
+        // array buffer only have long strings
+        let mut array_buffer: Vec<u8> = Vec::with_capacity(4096);
+
+        let buffer_id = output.buffers.len() as u32;
+
+        let read = if !self.validate_utf8 {
+            self.decoder.read(len, |bytes| {
+                let offset = array_buffer.len();
+                let view = make_view(bytes, buffer_id, offset as u32);
+                if bytes.len() > 12 {
+                    // only copy the data to buffer if the string can not be inlined.
+                    array_buffer.extend_from_slice(bytes);
+                }
+
+                // # Safety
+                // The buffer_id is the last buffer in the output buffers
+                // The offset is calculated from the buffer, so it is valid
+                unsafe {
+                    output.append_raw_view_unchecked(&view);
+                }
+                Ok(())
+            })?
+        } else {
+            // utf8 validation buffer has only short strings. These short
+            // strings are inlined into the views but we copy them into a
+            // contiguous buffer to accelerate validation.®
+            let mut utf8_validation_buffer = Vec::with_capacity(4096);
+
+            let v = self.decoder.read(len, |bytes| {
+                let offset = array_buffer.len();
+                let view = make_view(bytes, buffer_id, offset as u32);
+                if bytes.len() > 12 {
+                    // only copy the data to buffer if the string can not be inlined.
+                    array_buffer.extend_from_slice(bytes);
+                } else {
+                    utf8_validation_buffer.extend_from_slice(bytes);
+                }
+
+                // # Safety
+                // The buffer_id is the last buffer in the output buffers
+                // The offset is calculated from the buffer, so it is valid
+                // Utf-8 validation is done later
+                unsafe {
+                    output.append_raw_view_unchecked(&view);
+                }
+                Ok(())
+            })?;
+            check_valid_utf8(&array_buffer)?;
+            check_valid_utf8(&utf8_validation_buffer)?;
+            v
+        };
+
+        let actual_block_id = output.append_block(array_buffer.into());
+        assert_eq!(actual_block_id, buffer_id);
+        Ok(read)
+    }
+
+    fn skip(&mut self, to_skip: usize) -> Result<usize> {
+        self.decoder.skip(to_skip)
+    }
+}
+
 /// Check that `val` is a valid UTF-8 sequence
 pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
     match std::str::from_utf8(val) {
@@ -487,13 +709,6 @@ mod tests {
             .unwrap();
 
         for (encoding, page) in pages {
-            if encoding != Encoding::PLAIN
-                && encoding != Encoding::RLE_DICTIONARY
-                && encoding != Encoding::PLAIN_DICTIONARY
-            {
-                // skip unsupported encodings for now as they are not yet implemented
-                continue;
-            }
             let mut output = ViewBuffer::default();
             decoder.set_data(encoding, page, 4, Some(4)).unwrap();
 
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index b38092dbc..cc369cec0 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -2456,26 +2456,16 @@ mod tests {
         let cases = [
             (
                 invalid_utf8_first_char::<i32>(),
-                "Parquet argument error: Parquet error: encountered non UTF-8 data",
+                "Parquet argument error: Parquet error: encountered non UTF-8 data: invalid utf-8 sequence of 1 bytes from index 11",
             ),
             (
                 invalid_utf8_later_char::<i32>(),
-                "Parquet argument error: Parquet error: encountered non UTF-8 data: invalid utf-8 sequence of 1 bytes from index 6",
+                "Parquet argument error: Parquet error: encountered non UTF-8 data: invalid utf-8 sequence of 1 bytes from index 14",
             ),
         ];
         for (array, expected_error) in cases {
-            // cast not yet implemented for BinaryView
-            // https://github.com/apache/arrow-rs/issues/5508
-            // so copy directly
-            let mut builder = BinaryViewBuilder::with_capacity(100);
-            for v in array.iter() {
-                if let Some(v) = v {
-                    builder.append_value(v);
-                } else {
-                    builder.append_null();
-                }
-            }
-            let array = builder.finish();
+            let array = arrow_cast::cast(&array, &ArrowDataType::BinaryView).unwrap();
+            let array = array.as_binary_view();
 
             // data is not valid utf8 we can not construct a correct StringArray
             // safely, so purposely create an invalid StringArray
diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
index 0a72583b9..e4205b7ef 100644
--- a/parquet/src/arrow/async_reader/mod.rs
+++ b/parquet/src/arrow/async_reader/mod.rs
@@ -1650,7 +1650,7 @@ mod tests {
     #[tokio::test]
     async fn test_parquet_record_batch_stream_schema() {
         fn get_all_field_names(schema: &Schema) -> Vec<&String> {
-            schema.all_fields().iter().map(|f| f.name()).collect()
+            schema.flattened_fields().iter().map(|f| f.name()).collect()
         }
 
         // ParquetRecordBatchReaderBuilder::schema differs from
diff --git a/parquet/src/arrow/buffer/view_buffer.rs b/parquet/src/arrow/buffer/view_buffer.rs
index 1651aa2d7..ae83ac317 100644
--- a/parquet/src/arrow/buffer/view_buffer.rs
+++ b/parquet/src/arrow/buffer/view_buffer.rs
@@ -37,7 +37,6 @@ impl ViewBuffer {
         self.views.is_empty()
     }
 
-    #[allow(unused)]
     pub fn append_block(&mut self, block: Buffer) -> u32 {
         let block_id = self.buffers.len() as u32;
         self.buffers.push(block);
@@ -49,7 +48,6 @@ impl ViewBuffer {
     /// - `block` is a valid index, i.e., the return value of `append_block`
     /// - `offset` and `offset + len` are valid indices into the buffer
     /// - The `(offset, offset + len)` is valid value for the native type.
-    #[allow(unused)]
     pub unsafe fn append_view_unchecked(&mut self, block: u32, offset: u32, len: u32) {
         let b = self.buffers.get_unchecked(block as usize);
         let end = offset.saturating_add(len);


Test patch:
diff --git a/arrow-flight/tests/flight_sql_client_cli.rs b/arrow-flight/tests/flight_sql_client_cli.rs
index 50a4ec0d8..317eb3900 100644
--- a/arrow-flight/tests/flight_sql_client_cli.rs
+++ b/arrow-flight/tests/flight_sql_client_cli.rs
@@ -568,7 +568,7 @@ impl FlightSqlService for FlightSqlServiceImpl {
         .try_collect::<Vec<_>>()
         .await?;
 
-        for (left, right) in parameters[0].schema().all_fields().iter().zip(vec![
+        for (left, right) in parameters[0].schema().flattened_fields().iter().zip(vec![
             Field::new("$1", DataType::Utf8, false),
             Field::new("$2", DataType::Int64, true),
         ]) {