Repository: apache/arrow-rs
Pull request #6001 (2024-07-04 01:53:00)

Rename `Schema::all_fields` to `flattened_fields`
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes #.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

`Schema::all_fields` is very easy to be misunderstood and cause bug, I propose to rename it to `flattened_fields`.

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index dc4cbe683..a00bf7271 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -325,7 +325,8 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
     /// Use with caution as this can be an expensive operation, only use it when you are sure that the view
     /// array is significantly smaller than when it is originally created, e.g., after filtering or slicing.
     pub fn gc(&self) -> Self {
-        let mut builder = GenericByteViewBuilder::<T>::with_capacity(self.len());
+        let mut builder =
+            GenericByteViewBuilder::<T>::with_capacity(self.len()).with_deduplicate_strings();
 
         for v in self.iter() {
             builder.append_option(v);
diff --git a/arrow-array/src/builder/generic_bytes_view_builder.rs b/arrow-array/src/builder/generic_bytes_view_builder.rs
index 2bcc5a3f3..dda553545 100644
--- a/arrow-array/src/builder/generic_bytes_view_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_view_builder.rs
@@ -22,6 +22,8 @@ use std::sync::Arc;
 use arrow_buffer::{Buffer, BufferBuilder, NullBufferBuilder, ScalarBuffer};
 use arrow_data::ByteView;
 use arrow_schema::ArrowError;
+use hashbrown::hash_table::Entry;
+use hashbrown::HashTable;
 
 use crate::builder::ArrayBuilder;
 use crate::types::bytes::ByteArrayNativeType;
@@ -57,6 +59,9 @@ pub struct GenericByteViewBuilder<T: ByteViewType + ?Sized> {
     completed: Vec<Buffer>,
     in_progress: Vec<u8>,
     block_size: u32,
+    /// Some if deduplicating strings
+    /// map `<string hash> -> <index to the views>`
+    string_tracker: Option<(HashTable<usize>, ahash::RandomState)>,
     phantom: PhantomData<T>,
 }
 
@@ -74,6 +79,7 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
             completed: vec![],
             in_progress: vec![],
             block_size: DEFAULT_BLOCK_SIZE,
+            string_tracker: None,
             phantom: Default::default(),
         }
     }
@@ -83,6 +89,20 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
         Self { block_size, ..self }
     }
 
+    /// Deduplicate strings while building the array
+    ///
+    /// This will potentially decrease the memory usage if the array have repeated strings
+    /// It will also increase the time to build the array as it needs to hash the strings
+    pub fn with_deduplicate_strings(self) -> Self {
+        Self {
+            string_tracker: Some((
+                HashTable::with_capacity(self.views_builder.capacity()),
+                Default::default(),
+            )),
+            ..self
+        }
+    }
+
     /// Append a new data block returning the new block offset
     ///
     /// Note: this will first flush any in-progress block
@@ -179,6 +199,26 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
         self.completed.push(block);
     }
 
+    /// Returns the value at the given index
+    /// Useful if we want to know what value has been inserted to the builder
+    fn get_value(&self, index: usize) -> &[u8] {
+        let view = self.views_builder.as_slice().get(index).unwrap();
+        let len = *view as u32;
+        if len <= 12 {
+            // # Safety
+            // The view is valid from the builder
+            unsafe { GenericByteViewArray::<T>::inline_value(view, len as usize) }
+        } else {
+            let view = ByteView::from(*view);
+            if view.buffer_index < self.completed.len() as u32 {
+                let block = &self.completed[view.buffer_index as usize];
+                &block[view.offset as usize..view.offset as usize + view.length as usize]
+            } else {
+                &self.in_progress[view.offset as usize..view.offset as usize + view.length as usize]
+            }
+        }
+    }
+
     /// Appends a value into the builder
     ///
     /// # Panics
@@ -199,6 +239,40 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
             return;
         }
 
+        // Deduplication if:
+        // (1) deduplication is enabled.
+        // (2) len > 12
+        if let Some((mut ht, hasher)) = self.string_tracker.take() {
+            let hash_val = hasher.hash_one(v);
+            let hasher_fn = |v: &_| hasher.hash_one(v);
+
+            let entry = ht.entry(
+                hash_val,
+                |idx| {
+                    let stored_value = self.get_value(*idx);
+                    v == stored_value
+                },
+                hasher_fn,
+            );
+            match entry {
+                Entry::Occupied(occupied) => {
+                    // If the string already exists, we will directly use the view
+                    let idx = occupied.get();
+                    self.views_builder
+                        .append(self.views_builder.as_slice()[*idx]);
+                    self.null_buffer_builder.append_non_null();
+                    self.string_tracker = Some((ht, hasher));
+                    return;
+                }
+                Entry::Vacant(vacant) => {
+                    // o.w. we insert the (string hash -> view index)
+                    // the idx is current length of views_builder, as we are inserting a new view
+                    vacant.insert(self.views_builder.len());
+                }
+            }
+            self.string_tracker = Some((ht, hasher));
+        }
+
         let required_cap = self.in_progress.len() + v.len();
         if self.in_progress.capacity() < required_cap {
             self.flush_in_progress();
@@ -357,6 +431,42 @@ mod tests {
     use super::*;
     use crate::Array;
 
+    #[test]
+    fn test_string_view_deduplicate() {
+        let value_1 = "long string to test string view";
+        let value_2 = "not so similar string but long";
+
+        let mut builder = StringViewBuilder::new()
+            .with_deduplicate_strings()
+            .with_block_size(value_1.len() as u32 * 2); // so that we will have multiple buffers
+
+        let values = vec![
+            Some(value_1),
+            Some(value_2),
+            Some("short"),
+            Some(value_1),
+            None,
+            Some(value_2),
+            Some(value_1),
+        ];
+        builder.extend(values.clone());
+
+        let array = builder.finish_cloned();
+        array.to_data().validate_full().unwrap();
+        assert_eq!(array.data_buffers().len(), 1); // without duplication we would need 3 buffers.
+        let actual: Vec<_> = array.iter().collect();
+        assert_eq!(actual, values);
+
+        let view0 = array.views().first().unwrap();
+        let view3 = array.views().get(3).unwrap();
+        let view6 = array.views().get(6).unwrap();
+
+        assert_eq!(view0, view3);
+        assert_eq!(view0, view6);
+
+        assert_eq!(array.views().get(1), array.views().get(5));
+    }
+
     #[test]
     fn test_string_view() {
         let b1 = Buffer::from(b"world\xFFbananas\xF0\x9F\x98\x81");
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index c01f20e05..df5c17676 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -69,10 +69,14 @@ pub struct FlightSqlServiceClient<T> {
 impl FlightSqlServiceClient<Channel> {
     /// Creates a new FlightSql client that connects to a server over an arbitrary tonic `Channel`
     pub fn new(channel: Channel) -> Self {
-        let flight_client = FlightServiceClient::new(channel);
-        FlightSqlServiceClient {
+        Self::new_from_inner(FlightServiceClient::new(channel))
+    }
+
+    /// Creates a new higher level client with the provided lower level client
+    pub fn new_from_inner(inner: FlightServiceClient<Channel>) -> Self {
+        Self {
             token: None,
-            flight_client,
+            flight_client: inner,
             headers: HashMap::default(),
         }
     }
diff --git a/arrow-flight/src/sql/metadata/sql_info.rs b/arrow-flight/src/sql/metadata/sql_info.rs
index d4584f4a6..97304d3c8 100644
--- a/arrow-flight/src/sql/metadata/sql_info.rs
+++ b/arrow-flight/src/sql/metadata/sql_info.rs
@@ -148,7 +148,7 @@ impl SqlInfoName for u32 {
 /// *              int32_to_int32_list_map: map<key: int32, value: list<$data$: int32>>
 /// * >
 /// ```
-///[flightsql]: (https://github.com/apache/arrow/blob/f9324b79bf4fc1ec7e97b32e3cce16e75ef0f5e3/format/FlightSql.proto#L32-L43
+///[flightsql]: https://github.com/apache/arrow/blob/f9324b79bf4fc1ec7e97b32e3cce16e75ef0f5e3/format/FlightSql.proto#L32-L43
 ///[Union Spec]: https://arrow.apache.org/docs/format/Columnar.html#dense-union
 struct SqlInfoUnionBuilder {
     // Values for each child type
@@ -361,7 +361,7 @@ impl SqlInfoDataBuilder {
 
     /// Encode the contents of this list according to the [FlightSQL spec]
     ///
-    /// [FlightSQL spec]: (https://github.com/apache/arrow/blob/f9324b79bf4fc1ec7e97b32e3cce16e75ef0f5e3/format/FlightSql.proto#L32-L43
+    /// [FlightSQL spec]: https://github.com/apache/arrow/blob/f9324b79bf4fc1ec7e97b32e3cce16e75ef0f5e3/format/FlightSql.proto#L32-L43
     pub fn build(self) -> Result<SqlInfoData> {
         let mut name_builder = UInt32Builder::new();
         let mut value_builder = SqlInfoUnionBuilder::new();
diff --git a/arrow-flight/src/sql/mod.rs b/arrow-flight/src/sql/mod.rs
index 089ee4dd8..b3b9882ee 100644
--- a/arrow-flight/src/sql/mod.rs
+++ b/arrow-flight/src/sql/mod.rs
@@ -45,6 +45,7 @@ use prost::Message;
 
 mod gen {
     #![allow(clippy::all)]
+    #![allow(rustdoc::unportable_markdown)]
     include!("arrow.flight.protocol.sql.rs");
 }
 
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/datatype.rs b/arrow-schema/src/datatype.rs
index 780e9de79..38e50a17a 100644
--- a/arrow-schema/src/datatype.rs
+++ b/arrow-schema/src/datatype.rs
@@ -21,15 +21,64 @@ use std::sync::Arc;
 
 use crate::{ArrowError, Field, FieldRef, Fields, UnionFields};
 
-/// The set of datatypes that are supported by this implementation of Apache Arrow.
+/// Datatypes supported by this implementation of Apache Arrow.
 ///
-/// The Arrow specification on data types includes some more types.
-/// See also [`Schema.fbs`](https://github.com/apache/arrow/blob/main/format/Schema.fbs)
-/// for Arrow's specification.
+/// The variants of this enum include primitive fixed size types as well as
+/// parametric or nested types. See [`Schema.fbs`] for Arrow's specification.
 ///
-/// The variants of this enum include primitive fixed size types as well as parametric or
-/// nested types.
-/// Currently the Rust implementation supports the following nested types:
+/// # Examples
+///
+/// Primitive types
+/// ```
+/// # use arrow_schema::DataType;
+/// // create a new 32-bit signed integer
+/// let data_type = DataType::Int32;
+/// ```
+///
+/// Nested Types
+/// ```
+/// # use arrow_schema::{DataType, Field};
+/// # use std::sync::Arc;
+/// // create a new list of 32-bit signed integers directly
+/// let list_data_type = DataType::List(Arc::new(Field::new("item", DataType::Int32, true)));
+/// // Create the same list type with constructor
+/// let list_data_type2 = DataType::new_list(DataType::Int32, true);
+/// assert_eq!(list_data_type, list_data_type2);
+/// ```
+///
+/// Dictionary Types
+/// ```
+/// # use arrow_schema::{DataType};
+/// // String Dictionary (key type Int32 and value type Utf8)
+/// let data_type = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+/// ```
+///
+/// Timestamp Types
+/// ```
+/// # use arrow_schema::{DataType, TimeUnit};
+/// // timestamp with millisecond precision without timezone specified
+/// let data_type = DataType::Timestamp(TimeUnit::Millisecond, None);
+/// // timestamp with nanosecond precision in UTC timezone
+/// let data_type = DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()));
+///```
+///
+/// # Display and FromStr
+///
+/// The `Display` and `FromStr` implementations for `DataType` are
+/// human-readable, parseable, and reversible.
+///
+/// ```
+/// # use arrow_schema::DataType;
+/// let data_type = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+/// let data_type_string = data_type.to_string();
+/// assert_eq!(data_type_string, "Dictionary(Int32, Utf8)");
+/// // display can be parsed back into the original type
+/// let parsed_data_type: DataType = data_type.to_string().parse().unwrap();
+/// assert_eq!(data_type, parsed_data_type);
+/// ```
+///
+/// # Nested Support
+/// Currently, the Rust implementation supports the following nested types:
 ///  - `List<T>`
 ///  - `LargeList<T>`
 ///  - `FixedSizeList<T>`
@@ -39,7 +88,10 @@ use crate::{ArrowError, Field, FieldRef, Fields, UnionFields};
 ///
 /// Nested types can themselves be nested within other arrays.
 /// For more information on these types please see
-/// [the physical memory layout of Apache Arrow](https://arrow.apache.org/docs/format/Columnar.html#physical-memory-layout).
+/// [the physical memory layout of Apache Arrow]
+///
+/// [`Schema.fbs`]: https://github.com/apache/arrow/blob/main/format/Schema.fbs
+/// [the physical memory layout of Apache Arrow]: https://arrow.apache.org/docs/format/Columnar.html#physical-memory-layout
 #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub enum DataType {
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/object_store/src/local.rs b/object_store/src/local.rs
index d3bfab8ed..484738929 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -240,6 +240,8 @@ impl From<Error> for super::Error {
 #[derive(Debug)]
 pub struct LocalFileSystem {
     config: Arc<Config>,
+    // if you want to delete empty directories when deleting files
+    automatic_cleanup: bool,
 }
 
 #[derive(Debug)]
@@ -266,6 +268,7 @@ impl LocalFileSystem {
             config: Arc::new(Config {
                 root: Url::parse("file:///").unwrap(),
             }),
+            automatic_cleanup: false,
         }
     }
 
@@ -282,6 +285,7 @@ impl LocalFileSystem {
             config: Arc::new(Config {
                 root: absolute_path_to_url(path)?,
             }),
+            automatic_cleanup: false,
         })
     }
 
@@ -295,6 +299,12 @@ impl LocalFileSystem {
         );
         self.config.prefix_to_filesystem(location)
     }
+
+    /// Enable automatic cleanup of empty directories when deleting files
+    pub fn with_automatic_cleanup(mut self, automatic_cleanup: bool) -> Self {
+        self.automatic_cleanup = automatic_cleanup;
+        self
+    }
 }
 
 impl Config {
@@ -465,13 +475,36 @@ impl ObjectStore for LocalFileSystem {
     }
 
     async fn delete(&self, location: &Path) -> Result<()> {
+        let config = Arc::clone(&self.config);
         let path = self.path_to_filesystem(location)?;
-        maybe_spawn_blocking(move || match std::fs::remove_file(&path) {
-            Ok(_) => Ok(()),
-            Err(e) => Err(match e.kind() {
-                ErrorKind::NotFound => Error::NotFound { path, source: e }.into(),
-                _ => Error::UnableToDeleteFile { path, source: e }.into(),
-            }),
+        let automactic_cleanup = self.automatic_cleanup;
+        maybe_spawn_blocking(move || {
+            if let Err(e) = std::fs::remove_file(&path) {
+                Err(match e.kind() {
+                    ErrorKind::NotFound => Error::NotFound { path, source: e }.into(),
+                    _ => Error::UnableToDeleteFile { path, source: e }.into(),
+                })
+            } else if automactic_cleanup {
+                let root = &config.root;
+                let root = root
+                    .to_file_path()
+                    .map_err(|_| Error::InvalidUrl { url: root.clone() })?;
+
+                // here we will try to traverse up and delete an empty dir if possible until we reach the root or get an error
+                let mut parent = path.parent();
+
+                while let Some(loc) = parent {
+                    if loc != root && std::fs::remove_dir(loc).is_ok() {
+                        parent = loc.parent();
+                    } else {
+                        break;
+                    }
+                }
+
+                Ok(())
+            } else {
+                Ok(())
+            }
         })
         .await
     }
@@ -1010,6 +1043,8 @@ fn convert_walkdir_result(
 
 #[cfg(test)]
 mod tests {
+    use std::fs;
+
     use futures::TryStreamExt;
     use tempfile::{NamedTempFile, TempDir};
 
@@ -1445,6 +1480,34 @@ mod tests {
         list.sort_unstable();
         assert_eq!(list, vec![c, a]);
     }
+
+    #[tokio::test]
+    async fn delete_dirs_automatically() {
+        let root = TempDir::new().unwrap();
+        let integration = LocalFileSystem::new_with_prefix(root.path())
+            .unwrap()
+            .with_automatic_cleanup(true);
+        let location = Path::from("nested/file/test_file");
+        let data = Bytes::from("arbitrary data");
+
+        integration
+            .put(&location, data.clone().into())
+            .await
+            .unwrap();
+
+        let read_data = integration
+            .get(&location)
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+
+        assert_eq!(&*read_data, data);
+        assert!(fs::read_dir(root.path()).unwrap().count() > 0);
+        integration.delete(&location).await.unwrap();
+        assert!(fs::read_dir(root.path()).unwrap().count() == 0);
+    }
 }
 
 #[cfg(not(target_arch = "wasm32"))]
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 7f0a0dd2a..9c5caaad5 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -314,6 +314,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 +334,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 +375,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)
     }
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


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),
         ]) {