Repository: apache/arrow-rs
Pull request #5723 (2024-05-05 15:41:55)

Add `GenericListViewArray` for `ListView` & `LargeListView`
# 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.
-->

A part of https://github.com/apache/arrow-rs/issues/5501

# 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.
-->

# 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.
-->

Add GenericListViewArray implement

# 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.
-->


Kikkon: @tustvold  If you have time, I would appreciate it if you could help me review this PR. Thanks! 
Kikkon: Thank you for taking the time to review my code despite being busy and unwell. 
tustvold: For future reference, it is helpful for reviewers if you don't force push branches once reviews have been made, this makes it much easier to see what has changed
Kikkon: > For future reference, it is helpful for reviewers if you don't force push branches once reviews have been made, this makes it much easier to see what has changed. I'll try to find time to review this later today, but I now have to do a full review again which is quite time consuming on such a large PR

@tustvold  Thank you for your help. I will be more careful about the force push issue in my future commits. I am very sorry.
Kikkon: Hi @tustvold , perhaps when you're free this week, you could help me review this. This way, we can continue to move forward with this work. Thank you.
tustvold: I'm afraid I don't have capacity to assist with this initiative further, but perhaps @alamb may do
Kikkon: @alamb  Perhaps you can help advance this PR when they have time. 🥸
Kikkon: hi @westonpace please help me review again
alamb: I am depressed about the large review backlog in this crate. We are looking for more help from the community reviewing PRs -- see https://github.com/apache/arrow-rs/issues/6418 for more
Kikkon: @westonpace  😭 Apologies for the delay in processing this pull request due to work reasons. I hope you can help review it when you have time. 

Patch:
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index c53478d8b..b1b558057 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -44,8 +44,11 @@ use super::ByteArrayType;
 ///
 /// # See Also
 ///
-/// See [`StringViewArray`] for storing utf8 encoded string data and
-/// [`BinaryViewArray`] for storing bytes.
+/// * [`StringViewArray`] for storing utf8 encoded string data
+/// * [`BinaryViewArray`] for storing bytes
+/// * [`ByteView`] to interpret `u128`s layout of the views.
+///
+/// [`ByteView`]: arrow_data::ByteView
 ///
 /// # Notes
 ///
@@ -872,12 +875,9 @@ mod tests {
     #[should_panic(expected = "Invalid buffer index at 0: got index 3 but only has 1 buffers")]
     fn new_with_invalid_view_data() {
         let v = "large payload over 12 bytes";
-        let view = ByteView {
-            length: 13,
-            prefix: u32::from_le_bytes(v.as_bytes()[0..4].try_into().unwrap()),
-            buffer_index: 3,
-            offset: 1,
-        };
+        let view = ByteView::new(13, &v.as_bytes()[0..4])
+            .with_buffer_index(3)
+            .with_offset(1);
         let views = ScalarBuffer::from(vec![view.into()]);
         let buffers = vec![Buffer::from_slice_ref(v)];
         StringViewArray::new(views, buffers, None);
@@ -888,13 +888,12 @@ mod tests {
         expected = "Encountered non-UTF-8 data at index 0: invalid utf-8 sequence of 1 bytes from index 0"
     )]
     fn new_with_invalid_utf8_data() {
-        let v: Vec<u8> = vec![0xf0, 0x80, 0x80, 0x80];
-        let view = ByteView {
-            length: v.len() as u32,
-            prefix: u32::from_le_bytes(v[0..4].try_into().unwrap()),
-            buffer_index: 0,
-            offset: 0,
-        };
+        let v: Vec<u8> = vec![
+            // invalid UTF8
+            0xf0, 0x80, 0x80, 0x80, // more bytes to make it larger than 12
+            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+        ];
+        let view = ByteView::new(v.len() as u32, &v[0..4]);
         let views = ScalarBuffer::from(vec![view.into()]);
         let buffers = vec![Buffer::from_slice_ref(v)];
         StringViewArray::new(views, buffers, None);
diff --git a/arrow-array/src/array/list_view_array.rs b/arrow-array/src/array/list_view_array.rs
new file mode 100644
index 000000000..9b9078c23
--- /dev/null
+++ b/arrow-array/src/array/list_view_array.rs
@@ -0,0 +1,1035 @@
+// 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 arrow_buffer::{NullBuffer, ScalarBuffer};
+use arrow_data::{ArrayData, ArrayDataBuilder};
+use arrow_schema::{ArrowError, DataType, FieldRef};
+use std::any::Any;
+use std::ops::Add;
+use std::sync::Arc;
+
+use crate::array::{make_array, print_long_array};
+use crate::iterator::GenericListViewArrayIter;
+use crate::{new_empty_array, Array, ArrayAccessor, ArrayRef, FixedSizeListArray, OffsetSizeTrait};
+
+/// A [`GenericListViewArray`] of variable size lists, storing offsets as `i32`.
+pub type ListViewArray = GenericListViewArray<i32>;
+
+/// A [`GenericListViewArray`] of variable size lists, storing offsets as `i64`.
+pub type LargeListViewArray = GenericListViewArray<i64>;
+
+///
+/// Different from [`crate::GenericListArray`] as it stores both an offset and length
+/// meaning that take / filter operations can be implemented without copying the underlying data.
+///
+/// [Variable-size List Layout: ListView Layout]: https://arrow.apache.org/docs/format/Columnar.html#listview-layout
+#[derive(Clone)]
+pub struct GenericListViewArray<OffsetSize: OffsetSizeTrait> {
+    data_type: DataType,
+    nulls: Option<NullBuffer>,
+    values: ArrayRef,
+    value_offsets: ScalarBuffer<OffsetSize>,
+    value_sizes: ScalarBuffer<OffsetSize>,
+}
+
+impl<OffsetSize: OffsetSizeTrait> GenericListViewArray<OffsetSize> {
+    /// The data type constructor of listview array.
+    /// The input is the schema of the child array and
+    /// the output is the [`DataType`], ListView or LargeListView.
+    pub const DATA_TYPE_CONSTRUCTOR: fn(FieldRef) -> DataType = if OffsetSize::IS_LARGE {
+        DataType::LargeListView
+    } else {
+        DataType::ListView
+    };
+
+    /// Create a new [`GenericListViewArray`] from the provided parts
+    ///
+    /// # Errors
+    ///
+    /// Errors if
+    ///
+    /// * `offsets.len() != sizes.len()`
+    /// * `offsets.len() != nulls.len()`
+    /// * `offsets[i] > values.len()`
+    /// * `!field.is_nullable() && values.is_nullable()`
+    /// * `field.data_type() != values.data_type()`
+    /// * `0 <= offsets[i] <= length of the child array`
+    /// * `0 <= offsets[i] + size[i] <= length of the child array`
+    pub fn try_new(
+        field: FieldRef,
+        offsets: ScalarBuffer<OffsetSize>,
+        sizes: ScalarBuffer<OffsetSize>,
+        values: ArrayRef,
+        nulls: Option<NullBuffer>,
+    ) -> Result<Self, ArrowError> {
+        let len = offsets.len();
+        if let Some(n) = nulls.as_ref() {
+            if n.len() != len {
+                return Err(ArrowError::InvalidArgumentError(format!(
+                    "Incorrect length of null buffer for {}ListViewArray, expected {len} got {}",
+                    OffsetSize::PREFIX,
+                    n.len(),
+                )));
+            }
+        }
+        if len != sizes.len() {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "Length of offsets buffer and sizes buffer must be equal for {}ListViewArray, got {len} and {}",
+                OffsetSize::PREFIX, sizes.len()
+            )));
+        }
+
+        for (offset, size) in offsets.iter().zip(sizes.iter()) {
+            let offset = offset.as_usize();
+            let size = size.as_usize();
+            if offset.checked_add(size).ok_or_else(|| {
+                ArrowError::InvalidArgumentError(format!(
+                    "Overflow in offset + size for {}ListViewArray",
+                    OffsetSize::PREFIX
+                ))
+            })? > values.len()
+            {
+                return Err(ArrowError::InvalidArgumentError(format!(
+                    "Offset + size for {}ListViewArray must be within the bounds of the child array, got offset: {offset}, size: {size}, child array length: {}",
+                    OffsetSize::PREFIX,
+                    values.len()
+                )));
+            }
+        }
+
+        if !field.is_nullable() && values.is_nullable() {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "Non-nullable field of {}ListViewArray {:?} cannot contain nulls",
+                OffsetSize::PREFIX,
+                field.name()
+            )));
+        }
+
+        if field.data_type() != values.data_type() {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "{}ListViewArray expected data type {} got {} for {:?}",
+                OffsetSize::PREFIX,
+                field.data_type(),
+                values.data_type(),
+                field.name()
+            )));
+        }
+
+        Ok(Self {
+            data_type: Self::DATA_TYPE_CONSTRUCTOR(field),
+            nulls,
+            values,
+            value_offsets: offsets,
+            value_sizes: sizes,
+        })
+    }
+
+    /// Create a new [`GenericListViewArray`] from the provided parts
+    ///
+    /// # Panics
+    ///
+    /// Panics if [`Self::try_new`] returns an error
+    pub fn new(
+        field: FieldRef,
+        offsets: ScalarBuffer<OffsetSize>,
+        sizes: ScalarBuffer<OffsetSize>,
+        values: ArrayRef,
+        nulls: Option<NullBuffer>,
+    ) -> Self {
+        Self::try_new(field, offsets, sizes, values, nulls).unwrap()
+    }
+
+    /// Create a new [`GenericListViewArray`] of length `len` where all values are null
+    pub fn new_null(field: FieldRef, len: usize) -> Self {
+        let values = new_empty_array(field.data_type());
+        Self {
+            data_type: Self::DATA_TYPE_CONSTRUCTOR(field),
+            nulls: Some(NullBuffer::new_null(len)),
+            value_offsets: ScalarBuffer::from(vec![]),
+            value_sizes: ScalarBuffer::from(vec![]),
+            values,
+        }
+    }
+
+    /// Deconstruct this array into its constituent parts
+    pub fn into_parts(
+        self,
+    ) -> (
+        FieldRef,
+        ScalarBuffer<OffsetSize>,
+        ScalarBuffer<OffsetSize>,
+        ArrayRef,
+        Option<NullBuffer>,
+    ) {
+        let f = match self.data_type {
+            DataType::ListView(f) | DataType::LargeListView(f) => f,
+            _ => unreachable!(),
+        };
+        (
+            f,
+            self.value_offsets,
+            self.value_sizes,
+            self.values,
+            self.nulls,
+        )
+    }
+
+    /// Returns a reference to the offsets of this list
+    ///
+    /// Unlike [`Self::value_offsets`] this returns the [`ScalarBuffer`]
+    /// allowing for zero-copy cloning
+    #[inline]
+    pub fn offsets(&self) -> &ScalarBuffer<OffsetSize> {
+        &self.value_offsets
+    }
+
+    /// Returns a reference to the values of this list
+    #[inline]
+    pub fn values(&self) -> &ArrayRef {
+        &self.values
+    }
+
+    /// Returns a reference to the sizes of this list
+    ///
+    /// Unlike [`Self::value_sizes`] this returns the [`ScalarBuffer`]
+    /// allowing for zero-copy cloning
+    #[inline]
+    pub fn sizes(&self) -> &ScalarBuffer<OffsetSize> {
+        &self.value_sizes
+    }
+
+    /// Returns a clone of the value type of this list.
+    pub fn value_type(&self) -> DataType {
+        self.values.data_type().clone()
+    }
+
+    /// Returns ith value of this list view array.
+    /// # Safety
+    /// Caller must ensure that the index is within the array bounds
+    pub unsafe fn value_unchecked(&self, i: usize) -> ArrayRef {
+        let offset = self.value_offsets().get_unchecked(i).as_usize();
+        let length = self.value_sizes().get_unchecked(i).as_usize();
+        self.values.slice(offset, length)
+    }
+
+    /// Returns ith value of this list view array.
+    /// # Panics
+    /// Panics if the index is out of bounds
+    pub fn value(&self, i: usize) -> ArrayRef {
+        let offset = self.value_offsets()[i].as_usize();
+        let length = self.value_sizes()[i].as_usize();
+        self.values.slice(offset, length)
+    }
+
+    /// Returns the offset values in the offsets buffer
+    #[inline]
+    pub fn value_offsets(&self) -> &[OffsetSize] {
+        &self.value_offsets
+    }
+
+    /// Returns the sizes values in the offsets buffer
+    #[inline]
+    pub fn value_sizes(&self) -> &[OffsetSize] {
+        &self.value_sizes
+    }
+
+    /// Returns the size for value at index `i`.
+    #[inline]
+    pub fn value_size(&self, i: usize) -> OffsetSize {
+        self.value_sizes[i]
+    }
+
+    /// Returns the offset for value at index `i`.
+    pub fn value_offset(&self, i: usize) -> OffsetSize {
+        self.value_offsets[i]
+    }
+
+    /// Constructs a new iterator
+    pub fn iter(&self) -> GenericListViewArrayIter<'_, OffsetSize> {
+        GenericListViewArrayIter::<'_, OffsetSize>::new(self)
+    }
+
+    #[inline]
+    fn get_type(data_type: &DataType) -> Option<&DataType> {
+        match (OffsetSize::IS_LARGE, data_type) {
+            (true, DataType::LargeListView(child)) | (false, DataType::ListView(child)) => {
+                Some(child.data_type())
+            }
+            _ => None,
+        }
+    }
+
+    /// Returns a zero-copy slice of this array with the indicated offset and length.
+    pub fn slice(&self, offset: usize, length: usize) -> Self {
+        Self {
+            data_type: self.data_type.clone(),
+            nulls: self.nulls.as_ref().map(|n| n.slice(offset, length)),
+            values: self.values.clone(),
+            value_offsets: self.value_offsets.slice(offset, length),
+            value_sizes: self.value_sizes.slice(offset, length),
+        }
+    }
+}
+
+impl<'a, OffsetSize: OffsetSizeTrait> ArrayAccessor for &'a GenericListViewArray<OffsetSize> {
+    type Item = ArrayRef;
+
+    fn value(&self, index: usize) -> Self::Item {
+        GenericListViewArray::value(self, index)
+    }
+
+    unsafe fn value_unchecked(&self, index: usize) -> Self::Item {
+        GenericListViewArray::value_unchecked(self, index)
+    }
+}
+
+impl<OffsetSize: OffsetSizeTrait> Array for GenericListViewArray<OffsetSize> {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn to_data(&self) -> ArrayData {
+        self.clone().into()
+    }
+
+    fn into_data(self) -> ArrayData {
+        self.into()
+    }
+
+    fn data_type(&self) -> &DataType {
+        &self.data_type
+    }
+
+    fn slice(&self, offset: usize, length: usize) -> ArrayRef {
+        Arc::new(self.slice(offset, length))
+    }
+
+    fn len(&self) -> usize {
+        self.sizes().len()
+    }
+
+    fn is_empty(&self) -> bool {
+        self.value_sizes.is_empty()
+    }
+
+    fn offset(&self) -> usize {
+        0
+    }
+
+    fn nulls(&self) -> Option<&NullBuffer> {
+        self.nulls.as_ref()
+    }
+
+    fn get_buffer_memory_size(&self) -> usize {
+        let mut size = self.values.get_buffer_memory_size();
+        size += self.value_offsets.inner().capacity();
+        size += self.value_sizes.inner().capacity();
+        if let Some(n) = self.nulls.as_ref() {
+            size += n.buffer().capacity();
+        }
+        size
+    }
+
+    fn get_array_memory_size(&self) -> usize {
+        let mut size = std::mem::size_of::<Self>() + self.values.get_array_memory_size();
+        size += self.value_offsets.inner().capacity();
+        size += self.value_sizes.inner().capacity();
+        if let Some(n) = self.nulls.as_ref() {
+            size += n.buffer().capacity();
+        }
+        size
+    }
+}
+
+impl<OffsetSize: OffsetSizeTrait> std::fmt::Debug for GenericListViewArray<OffsetSize> {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        let prefix = OffsetSize::PREFIX;
+        write!(f, "{prefix}ListViewArray\n[\n")?;
+        print_long_array(self, f, |array, index, f| {
+            std::fmt::Debug::fmt(&array.value(index), f)
+        })?;
+        write!(f, "]")
+    }
+}
+
+impl<OffsetSize: OffsetSizeTrait> From<GenericListViewArray<OffsetSize>> for ArrayData {
+    fn from(array: GenericListViewArray<OffsetSize>) -> Self {
+        let len = array.len();
+        let builder = ArrayDataBuilder::new(array.data_type)
+            .len(len)
+            .nulls(array.nulls)
+            .buffers(vec![
+                array.value_offsets.into_inner(),
+                array.value_sizes.into_inner(),
+            ])
+            .child_data(vec![array.values.to_data()]);
+
+        unsafe { builder.build_unchecked() }
+    }
+}
+
+impl<OffsetSize: OffsetSizeTrait> From<ArrayData> for GenericListViewArray<OffsetSize> {
+    fn from(data: ArrayData) -> Self {
+        Self::try_new_from_array_data(data)
+            .expect("Expected infallible creation of GenericListViewArray from ArrayDataRef failed")
+    }
+}
+
+impl<OffsetSize: OffsetSizeTrait> From<FixedSizeListArray> for GenericListViewArray<OffsetSize> {
+    fn from(value: FixedSizeListArray) -> Self {
+        let (field, size) = match value.data_type() {
+            DataType::FixedSizeList(f, size) => (f, *size as usize),
+            _ => unreachable!(),
+        };
+        let mut acc = 0_usize;
+        let iter = std::iter::repeat(size).take(value.len());
+        let mut sizes = Vec::with_capacity(iter.size_hint().0);
+        let mut offsets = Vec::with_capacity(iter.size_hint().0);
+
+        for size in iter {
+            offsets.push(OffsetSize::usize_as(acc));
+            acc = acc.add(size);
+            sizes.push(OffsetSize::usize_as(size));
+        }
+        let sizes = ScalarBuffer::from(sizes);
+        let offsets = ScalarBuffer::from(offsets);
+        Self {
+            data_type: Self::DATA_TYPE_CONSTRUCTOR(field.clone()),
+            nulls: value.nulls().cloned(),
+            values: value.values().clone(),
+            value_offsets: offsets,
+            value_sizes: sizes,
+        }
+    }
+}
+
+impl<OffsetSize: OffsetSizeTrait> GenericListViewArray<OffsetSize> {
+    fn try_new_from_array_data(data: ArrayData) -> Result<Self, ArrowError> {
+        if data.buffers().len() != 2 {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "ListViewArray data should contain two buffers (value offsets & value sizes), had {}",
+                data.buffers().len()
+            )));
+        }
+
+        if data.child_data().len() != 1 {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "ListViewArray should contain a single child array (values array), had {}",
+                data.child_data().len()
+            )));
+        }
+
+        let values = data.child_data()[0].clone();
+
+        if let Some(child_data_type) = Self::get_type(data.data_type()) {
+            if values.data_type() != child_data_type {
+                return Err(ArrowError::InvalidArgumentError(format!(
+                    "{}ListViewArray's child datatype {:?} does not \
+                             correspond to the List's datatype {:?}",
+                    OffsetSize::PREFIX,
+                    values.data_type(),
+                    child_data_type
+                )));
+            }
+        } else {
+            return Err(ArrowError::InvalidArgumentError(format!(
+                "{}ListViewArray's datatype must be {}ListViewArray(). It is {:?}",
+                OffsetSize::PREFIX,
+                OffsetSize::PREFIX,
+                data.data_type()
+            )));
+        }
+
+        let values = make_array(values);
+        // ArrayData is valid, and verified type above
+        let value_offsets = ScalarBuffer::new(data.buffers()[0].clone(), data.offset(), data.len());
+        let value_sizes = ScalarBuffer::new(data.buffers()[1].clone(), data.offset(), data.len());
+
+        Ok(Self {
+            data_type: data.data_type().clone(),
+            nulls: data.nulls().cloned(),
+            values,
+            value_offsets,
+            value_sizes,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow_buffer::{bit_util, BooleanBuffer, Buffer, ScalarBuffer};
+    use arrow_schema::Field;
+
+    use crate::builder::{FixedSizeListBuilder, Int32Builder};
+    use crate::cast::AsArray;
+    use crate::types::Int32Type;
+    use crate::{Int32Array, Int64Array};
+
+    use super::*;
+
+    #[test]
+    fn test_empty_list_view_array() {
+        // Construct an empty value array
+        let vec: Vec<i32> = vec![];
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![]);
+        let offsets = ScalarBuffer::from(vec![]);
+        let values = Int32Array::from(vec);
+        let list_array = LargeListViewArray::new(field, offsets, sizes, Arc::new(values), None);
+
+        assert_eq!(list_array.len(), 0)
+    }
+
+    #[test]
+    fn test_list_view_array() {
+        // Construct a value array
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(8)
+            .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7]))
+            .build()
+            .unwrap();
+
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![3i32, 3, 2]);
+        let offsets = ScalarBuffer::from(vec![0i32, 3, 6]);
+        let values = Int32Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7]);
+        let list_array = ListViewArray::new(field, offsets, sizes, Arc::new(values), None);
+
+        let values = list_array.values();
+        assert_eq!(value_data, values.to_data());
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(3, list_array.len());
+        assert_eq!(0, list_array.null_count());
+        assert_eq!(6, list_array.value_offsets()[2]);
+        assert_eq!(2, list_array.value_sizes()[2]);
+        assert_eq!(2, list_array.value_size(2));
+        assert_eq!(0, list_array.value(0).as_primitive::<Int32Type>().value(0));
+        assert_eq!(
+            0,
+            unsafe { list_array.value_unchecked(0) }
+                .as_primitive::<Int32Type>()
+                .value(0)
+        );
+        for i in 0..3 {
+            assert!(list_array.is_valid(i));
+            assert!(!list_array.is_null(i));
+        }
+    }
+
+    #[test]
+    fn test_large_list_view_array() {
+        // Construct a value array
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(8)
+            .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7]))
+            .build()
+            .unwrap();
+
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![3i64, 3, 2]);
+        let offsets = ScalarBuffer::from(vec![0i64, 3, 6]);
+        let values = Int32Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7]);
+        let list_array = LargeListViewArray::new(field, offsets, sizes, Arc::new(values), None);
+
+        let values = list_array.values();
+        assert_eq!(value_data, values.to_data());
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(3, list_array.len());
+        assert_eq!(0, list_array.null_count());
+        assert_eq!(6, list_array.value_offsets()[2]);
+        assert_eq!(2, list_array.value_sizes()[2]);
+        assert_eq!(2, list_array.value_size(2));
+        assert_eq!(0, list_array.value(0).as_primitive::<Int32Type>().value(0));
+        assert_eq!(
+            0,
+            unsafe { list_array.value_unchecked(0) }
+                .as_primitive::<Int32Type>()
+                .value(0)
+        );
+        for i in 0..3 {
+            assert!(list_array.is_valid(i));
+            assert!(!list_array.is_null(i));
+        }
+    }
+
+    #[test]
+    fn test_list_view_array_slice() {
+        // Construct a value array
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(10)
+            .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]))
+            .build()
+            .unwrap();
+
+        // 01011001 00000001
+        let mut null_bits: [u8; 2] = [0; 2];
+        bit_util::set_bit(&mut null_bits, 0);
+        bit_util::set_bit(&mut null_bits, 3);
+        bit_util::set_bit(&mut null_bits, 4);
+        bit_util::set_bit(&mut null_bits, 6);
+        bit_util::set_bit(&mut null_bits, 8);
+        let buffer = BooleanBuffer::new(Buffer::from(null_bits), 0, 9);
+        let null_buffer = NullBuffer::new(buffer);
+
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![2, 0, 0, 2, 2, 0, 3, 0, 1]);
+        let offsets = ScalarBuffer::from(vec![0, 2, 2, 2, 4, 6, 6, 9, 9]);
+        let values = Int32Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+        let list_array =
+            ListViewArray::new(field, offsets, sizes, Arc::new(values), Some(null_buffer));
+
+        let values = list_array.values();
+        assert_eq!(value_data, values.to_data());
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(9, list_array.len());
+        assert_eq!(4, list_array.null_count());
+        assert_eq!(2, list_array.value_offsets()[3]);
+        assert_eq!(2, list_array.value_sizes()[3]);
+        assert_eq!(2, list_array.value_size(3));
+
+        let sliced_array = list_array.slice(1, 6);
+        assert_eq!(6, sliced_array.len());
+        assert_eq!(3, sliced_array.null_count());
+
+        for i in 0..sliced_array.len() {
+            if bit_util::get_bit(&null_bits, 1 + i) {
+                assert!(sliced_array.is_valid(i));
+            } else {
+                assert!(sliced_array.is_null(i));
+            }
+        }
+
+        // Check offset and length for each non-null value.
+        let sliced_list_array = sliced_array
+            .as_any()
+            .downcast_ref::<ListViewArray>()
+            .unwrap();
+        assert_eq!(2, sliced_list_array.value_offsets()[2]);
+        assert_eq!(2, sliced_list_array.value_sizes()[2]);
+        assert_eq!(2, sliced_list_array.value_size(2));
+
+        assert_eq!(4, sliced_list_array.value_offsets()[3]);
+        assert_eq!(2, sliced_list_array.value_sizes()[3]);
+        assert_eq!(2, sliced_list_array.value_size(3));
+
+        assert_eq!(6, sliced_list_array.value_offsets()[5]);
+        assert_eq!(3, sliced_list_array.value_sizes()[5]);
+        assert_eq!(3, sliced_list_array.value_size(5));
+    }
+
+    #[test]
+    fn test_large_list_view_array_slice() {
+        // Construct a value array
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(10)
+            .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]))
+            .build()
+            .unwrap();
+
+        // 01011001 00000001
+        let mut null_bits: [u8; 2] = [0; 2];
+        bit_util::set_bit(&mut null_bits, 0);
+        bit_util::set_bit(&mut null_bits, 3);
+        bit_util::set_bit(&mut null_bits, 4);
+        bit_util::set_bit(&mut null_bits, 6);
+        bit_util::set_bit(&mut null_bits, 8);
+        let buffer = BooleanBuffer::new(Buffer::from(null_bits), 0, 9);
+        let null_buffer = NullBuffer::new(buffer);
+
+        // Construct a large list view array from the above two
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![2i64, 0, 0, 2, 2, 0, 3, 0, 1]);
+        let offsets = ScalarBuffer::from(vec![0i64, 2, 2, 2, 4, 6, 6, 9, 9]);
+        let values = Int32Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+        let list_array =
+            LargeListViewArray::new(field, offsets, sizes, Arc::new(values), Some(null_buffer));
+
+        let values = list_array.values();
+        assert_eq!(value_data, values.to_data());
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(9, list_array.len());
+        assert_eq!(4, list_array.null_count());
+        assert_eq!(2, list_array.value_offsets()[3]);
+        assert_eq!(2, list_array.value_sizes()[3]);
+        assert_eq!(2, list_array.value_size(3));
+
+        let sliced_array = list_array.slice(1, 6);
+        assert_eq!(6, sliced_array.len());
+        assert_eq!(3, sliced_array.null_count());
+
+        for i in 0..sliced_array.len() {
+            if bit_util::get_bit(&null_bits, 1 + i) {
+                assert!(sliced_array.is_valid(i));
+            } else {
+                assert!(sliced_array.is_null(i));
+            }
+        }
+
+        // Check offset and length for each non-null value.
+        let sliced_list_array = sliced_array
+            .as_any()
+            .downcast_ref::<LargeListViewArray>()
+            .unwrap();
+        assert_eq!(2, sliced_list_array.value_offsets()[2]);
+        assert_eq!(2, sliced_list_array.value_size(2));
+        assert_eq!(2, sliced_list_array.value_sizes()[2]);
+
+        assert_eq!(4, sliced_list_array.value_offsets()[3]);
+        assert_eq!(2, sliced_list_array.value_size(3));
+        assert_eq!(2, sliced_list_array.value_sizes()[3]);
+
+        assert_eq!(6, sliced_list_array.value_offsets()[5]);
+        assert_eq!(3, sliced_list_array.value_size(5));
+        assert_eq!(2, sliced_list_array.value_sizes()[3]);
+    }
+
+    #[test]
+    #[should_panic(expected = "index out of bounds: the len is 9 but the index is 10")]
+    fn test_list_view_array_index_out_of_bound() {
+        // 01011001 00000001
+        let mut null_bits: [u8; 2] = [0; 2];
+        bit_util::set_bit(&mut null_bits, 0);
+        bit_util::set_bit(&mut null_bits, 3);
+        bit_util::set_bit(&mut null_bits, 4);
+        bit_util::set_bit(&mut null_bits, 6);
+        bit_util::set_bit(&mut null_bits, 8);
+        let buffer = BooleanBuffer::new(Buffer::from(null_bits), 0, 9);
+        let null_buffer = NullBuffer::new(buffer);
+
+        // Construct a buffer for value offsets, for the nested array:
+        //  [[0, 1], null, null, [2, 3], [4, 5], null, [6, 7, 8], null, [9]]
+        // Construct a list array from the above two
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![2i32, 0, 0, 2, 2, 0, 3, 0, 1]);
+        let offsets = ScalarBuffer::from(vec![0i32, 2, 2, 2, 4, 6, 6, 9, 9]);
+        let values = Int32Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+        let list_array =
+            ListViewArray::new(field, offsets, sizes, Arc::new(values), Some(null_buffer));
+
+        assert_eq!(9, list_array.len());
+        list_array.value(10);
+    }
+    #[test]
+    #[should_panic(
+        expected = "ListViewArray data should contain two buffers (value offsets & value sizes), had 0"
+    )]
+    #[cfg(not(feature = "force_validate"))]
+    fn test_list_view_array_invalid_buffer_len() {
+        let value_data = unsafe {
+            ArrayData::builder(DataType::Int32)
+                .len(8)
+                .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7]))
+                .build_unchecked()
+        };
+        let list_data_type =
+            DataType::ListView(Arc::new(Field::new("item", DataType::Int32, false)));
+        let list_data = unsafe {
+            ArrayData::builder(list_data_type)
+                .len(3)
+                .add_child_data(value_data)
+                .build_unchecked()
+        };
+        drop(ListViewArray::from(list_data));
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "ListViewArray data should contain two buffers (value offsets & value sizes), had 1"
+    )]
+    #[cfg(not(feature = "force_validate"))]
+    fn test_list_view_array_invalid_child_array_len() {
+        let value_offsets = Buffer::from_slice_ref([0, 2, 5, 7]);
+        let list_data_type =
+            DataType::ListView(Arc::new(Field::new("item", DataType::Int32, false)));
+        let list_data = unsafe {
+            ArrayData::builder(list_data_type)
+                .len(3)
+                .add_buffer(value_offsets)
+                .build_unchecked()
+        };
+        drop(ListViewArray::from(list_data));
+    }
+
+    #[test]
+    fn test_list_view_array_offsets_need_not_start_at_zero() {
+        let field = Arc::new(Field::new("item", DataType::Int32, true));
+        let sizes = ScalarBuffer::from(vec![0i32, 0, 3]);
+        let offsets = ScalarBuffer::from(vec![2i32, 2, 5]);
+        let values = Int32Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7]);
+        let list_array = ListViewArray::new(field, offsets, sizes, Arc::new(values), None);
+
+        assert_eq!(list_array.value_size(0), 0);
+        assert_eq!(list_array.value_size(1), 0);
+        assert_eq!(list_array.value_size(2), 3);
+    }
+
+    #[test]
+    #[should_panic(expected = "Memory pointer is not aligned with the specified scalar type")]
+    #[cfg(not(feature = "force_validate"))]
+    fn test_list_view_array_alignment() {
+        let offset_buf = Buffer::from_slice_ref([0_u64]);
+        let offset_buf2 = offset_buf.slice(1);
+
+        let size_buf = Buffer::from_slice_ref([0_u64]);
+        let size_buf2 = size_buf.slice(1);
+
+        let values: [i32; 8] = [0; 8];
+        let value_data = unsafe {
+            ArrayData::builder(DataType::Int32)
+                .add_buffer(Buffer::from_slice_ref(values))
+                .build_unchecked()
+        };
+
+        let list_data_type =
+            DataType::ListView(Arc::new(Field::new("item", DataType::Int32, false)));
+        let list_data = unsafe {
+            ArrayData::builder(list_data_type)
+                .add_buffer(offset_buf2)
+                .add_buffer(size_buf2)
+                .add_child_data(value_data)
+                .build_unchecked()
+        };
+        drop(ListViewArray::from(list_data));
+    }
+
+    #[test]
+    fn test_empty_offsets() {
+        let f = Arc::new(Field::new("element", DataType::Int32, true));
+        let string = ListViewArray::from(
+            ArrayData::builder(DataType::ListView(f.clone()))
+                .buffers(vec![Buffer::from(&[]), Buffer::from(&[])])
+                .add_child_data(ArrayData::new_empty(&DataType::Int32))
+                .build()
+                .unwrap(),
+        );
+        assert_eq!(string.value_offsets(), &[]);
+        assert_eq!(string.value_sizes(), &[]);
+
+        let string = LargeListViewArray::from(
+            ArrayData::builder(DataType::LargeListView(f))
+                .buffers(vec![Buffer::from(&[]), Buffer::from(&[])])
+                .add_child_data(ArrayData::new_empty(&DataType::Int32))
+                .build()
+                .unwrap(),
+        );
+        assert_eq!(string.len(), 0);
+        assert_eq!(string.value_offsets(), &[]);
+        assert_eq!(string.value_sizes(), &[]);
+    }
+
+    #[test]
+    fn test_try_new() {
+        let offsets = ScalarBuffer::from(vec![0, 1, 4, 5]);
+        let sizes = ScalarBuffer::from(vec![1, 3, 1, 0]);
+        let values = Int32Array::new(vec![1, 2, 3, 4, 5].into(), None);
+        let values = Arc::new(values) as ArrayRef;
+
+        let field = Arc::new(Field::new("element", DataType::Int32, false));
+        ListViewArray::new(
+            field.clone(),
+            offsets.clone(),
+            sizes.clone(),
+            values.clone(),
+            None,
+        );
+
+        let nulls = NullBuffer::new_null(4);
+        ListViewArray::new(
+            field.clone(),
+            offsets,
+            sizes.clone(),
+            values.clone(),
+            Some(nulls),
+        );
+
+        let nulls = NullBuffer::new_null(4);
+        let offsets = ScalarBuffer::from(vec![0, 1, 2, 3, 4]);
+        let sizes = ScalarBuffer::from(vec![1, 1, 1, 1, 0]);
+        let err = LargeListViewArray::try_new(
+            field,
+            offsets.clone(),
+            sizes.clone(),
+            values.clone(),
+            Some(nulls),
+        )
+        .unwrap_err();
+
+        assert_eq!(
+            err.to_string(),
+            "Invalid argument error: Incorrect length of null buffer for LargeListViewArray, expected 5 got 4"
+        );
+
+        let field = Arc::new(Field::new("element", DataType::Int64, false));
+        let err = LargeListViewArray::try_new(
+            field.clone(),
+            offsets.clone(),
+            sizes.clone(),
+            values.clone(),
+            None,
+        )
+        .unwrap_err();
+
+        assert_eq!(
+            err.to_string(),
+            "Invalid argument error: LargeListViewArray expected data type Int64 got Int32 for \"element\""
+        );
+
+        let nulls = NullBuffer::new_null(7);
+        let values = Int64Array::new(vec![0; 7].into(), Some(nulls));
+        let values = Arc::new(values);
+
+        let err = LargeListViewArray::try_new(
+            field,
+            offsets.clone(),
+            sizes.clone(),
+            values.clone(),
+            None,
+        )
+        .unwrap_err();
+
+        assert_eq!(
+            err.to_string(),
+            "Invalid argument error: Non-nullable field of LargeListViewArray \"element\" cannot contain nulls"
+        );
+    }
+
+    #[test]
+    fn test_from_fixed_size_list() {
+        let mut builder = FixedSizeListBuilder::new(Int32Builder::new(), 3);
+        builder.values().append_slice(&[1, 2, 3]);
+        builder.append(true);
+        builder.values().append_slice(&[0, 0, 0]);
+        builder.append(false);
+        builder.values().append_slice(&[4, 5, 6]);
+        builder.append(true);
+        let list: ListViewArray = builder.finish().into();
+        let values: Vec<_> = list
+            .iter()
+            .map(|x| x.map(|x| x.as_primitive::<Int32Type>().values().to_vec()))
+            .collect();
+        assert_eq!(values, vec![Some(vec![1, 2, 3]), None, Some(vec![4, 5, 6])]);
+        let offsets = list.value_offsets();
+        assert_eq!(offsets, &[0, 3, 6]);
+        let sizes = list.value_sizes();
+        assert_eq!(sizes, &[3, 3, 3]);
+    }
+
+    #[test]
+    fn test_list_view_array_overlap_lists() {
+        let value_data = unsafe {
+            ArrayData::builder(DataType::Int32)
+                .len(8)
+                .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7]))
+                .build_unchecked()
+        };
+        let list_data_type =
+            DataType::ListView(Arc::new(Field::new("item", DataType::Int32, false)));
+        let list_data = unsafe {
+            ArrayData::builder(list_data_type)
+                .len(2)
+                .add_buffer(Buffer::from_slice_ref([0, 3])) // offsets
+                .add_buffer(Buffer::from_slice_ref([5, 5])) // sizes
+                .add_child_data(value_data)
+                .build_unchecked()
+        };
+        let array = ListViewArray::from(list_data);
+
+        assert_eq!(array.len(), 2);
+        assert_eq!(array.value_size(0), 5);
+        assert_eq!(array.value_size(1), 5);
+
+        let values: Vec<_> = array
+            .iter()
+            .map(|x| x.map(|x| x.as_primitive::<Int32Type>().values().to_vec()))
+            .collect();
+        assert_eq!(
+            values,
+            vec![Some(vec![0, 1, 2, 3, 4]), Some(vec![3, 4, 5, 6, 7])]
+        );
+    }
+
+    #[test]
+    fn test_list_view_array_incomplete_offsets() {
+        let value_data = unsafe {
+            ArrayData::builder(DataType::Int32)
+                .len(50)
+                .add_buffer(Buffer::from_slice_ref((0..50).collect::<Vec<i32>>()))
+                .build_unchecked()
+        };
+        let list_data_type =
+            DataType::ListView(Arc::new(Field::new("item", DataType::Int32, false)));
+        let list_data = unsafe {
+            ArrayData::builder(list_data_type)
+                .len(3)
+                .add_buffer(Buffer::from_slice_ref([0, 5, 10])) // offsets
+                .add_buffer(Buffer::from_slice_ref([0, 5, 10])) // sizes
+                .add_child_data(value_data)
+                .build_unchecked()
+        };
+        let array = ListViewArray::from(list_data);
+
+        assert_eq!(array.len(), 3);
+        assert_eq!(array.value_size(0), 0);
+        assert_eq!(array.value_size(1), 5);
+        assert_eq!(array.value_size(2), 10);
+
+        let values: Vec<_> = array
+            .iter()
+            .map(|x| x.map(|x| x.as_primitive::<Int32Type>().values().to_vec()))
+            .collect();
+        assert_eq!(
+            values,
+            vec![
+                Some(vec![]),
+                Some(vec![5, 6, 7, 8, 9]),
+                Some(vec![10, 11, 12, 13, 14, 15, 16, 17, 18, 19])
+            ]
+        );
+    }
+
+    #[test]
+    fn test_list_view_array_empty_lists() {
+        let value_data = unsafe {
+            ArrayData::builder(DataType::Int32)
+                .len(0)
+                .add_buffer(Buffer::from_slice_ref::<i32, &[_; 0]>(&[]))
+                .build_unchecked()
+        };
+        let list_data_type =
+            DataType::ListView(Arc::new(Field::new("item", DataType::Int32, false)));
+        let list_data = unsafe {
+            ArrayData::builder(list_data_type)
+                .len(3)
+                .add_buffer(Buffer::from_slice_ref([0, 0, 0])) // offsets
+                .add_buffer(Buffer::from_slice_ref([0, 0, 0])) // sizes
+                .add_child_data(value_data)
+                .build_unchecked()
+        };
+        let array = ListViewArray::from(list_data);
+
+        assert_eq!(array.len(), 3);
+        assert_eq!(array.value_size(0), 0);
+        assert_eq!(array.value_size(1), 0);
+        assert_eq!(array.value_size(2), 0);
+
+        let values: Vec<_> = array
+            .iter()
+            .map(|x| x.map(|x| x.as_primitive::<Int32Type>().values().to_vec()))
+            .collect();
+        assert_eq!(values, vec![Some(vec![]), Some(vec![]), Some(vec![])]);
+    }
+}
diff --git a/arrow-array/src/array/mod.rs b/arrow-array/src/array/mod.rs
index 50c5699ba..4b426e7ee 100644
--- a/arrow-array/src/array/mod.rs
+++ b/arrow-array/src/array/mod.rs
@@ -72,6 +72,10 @@ mod byte_view_array;
 
 pub use byte_view_array::*;
 
+mod list_view_array;
+
+pub use list_view_array::*;
+
 /// An array in the [arrow columnar format](https://arrow.apache.org/docs/format/Columnar.html)
 pub trait Array: std::fmt::Debug + Send + Sync {
     /// Returns the array as [`Any`] so that it can be
@@ -590,6 +594,12 @@ impl<OffsetSize: OffsetSizeTrait> PartialEq for GenericListArray<OffsetSize> {
     }
 }
 
+impl<OffsetSize: OffsetSizeTrait> PartialEq for GenericListViewArray<OffsetSize> {
+    fn eq(&self, other: &Self) -> bool {
+        self.to_data().eq(&other.to_data())
+    }
+}
+
 impl PartialEq for MapArray {
     fn eq(&self, other: &Self) -> bool {
         self.to_data().eq(&other.to_data())
@@ -678,6 +688,8 @@ pub fn make_array(data: ArrayData) -> ArrayRef {
         DataType::Utf8View => Arc::new(StringViewArray::from(data)) as ArrayRef,
         DataType::List(_) => Arc::new(ListArray::from(data)) as ArrayRef,
         DataType::LargeList(_) => Arc::new(LargeListArray::from(data)) as ArrayRef,
+        DataType::ListView(_) => Arc::new(ListViewArray::from(data)) as ArrayRef,
+        DataType::LargeListView(_) => Arc::new(LargeListViewArray::from(data)) as ArrayRef,
         DataType::Struct(_) => Arc::new(StructArray::from(data)) as ArrayRef,
         DataType::Map(_, _) => Arc::new(MapArray::from(data)) as ArrayRef,
         DataType::Union(_, _) => Arc::new(UnionArray::from(data)) as ArrayRef,
diff --git a/arrow-array/src/iterator.rs b/arrow-array/src/iterator.rs
index 3f9cc0d52..6708da3d5 100644
--- a/arrow-array/src/iterator.rs
+++ b/arrow-array/src/iterator.rs
@@ -21,7 +21,7 @@ use crate::array::{
     ArrayAccessor, BooleanArray, FixedSizeBinaryArray, GenericBinaryArray, GenericListArray,
     GenericStringArray, PrimitiveArray,
 };
-use crate::{FixedSizeListArray, MapArray};
+use crate::{FixedSizeListArray, GenericListViewArray, MapArray};
 use arrow_buffer::NullBuffer;
 
 /// An iterator that returns Some(T) or None, that can be used on any [`ArrayAccessor`]
@@ -143,7 +143,8 @@ pub type FixedSizeListIter<'a> = ArrayIter<&'a FixedSizeListArray>;
 pub type GenericListArrayIter<'a, O> = ArrayIter<&'a GenericListArray<O>>;
 /// an iterator that returns Some(T) or None, that can be used on any MapArray
 pub type MapArrayIter<'a> = ArrayIter<&'a MapArray>;
-
+/// an iterator that returns Some(T) or None, that can be used on any ListArray
+pub type GenericListViewArrayIter<'a, O> = ArrayIter<&'a GenericListViewArray<O>>;
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
diff --git a/arrow-data/src/byte_view.rs b/arrow-data/src/byte_view.rs
index 6f6d6d175..3b3ec6246 100644
--- a/arrow-data/src/byte_view.rs
+++ b/arrow-data/src/byte_view.rs
@@ -21,8 +21,40 @@ use arrow_schema::ArrowError;
 /// Helper to access views of [`GenericByteViewArray`] (`StringViewArray` and
 /// `BinaryViewArray`) where the length is greater than 12 bytes.
 ///
-/// See the documentation on [`GenericByteViewArray`] for more information on
-/// the layout of the views.
+/// See Also:
+/// * [`GenericByteViewArray`] for more information on the layout of the views.
+/// * [`validate_binary_view`] and [`validate_string_view`] to validate
+///
+/// # Example: Create a new u128 view
+///
+/// ```rust
+/// # use arrow_data::ByteView;;
+/// // Create a view for a string of length 20
+/// // first four bytes are "Rust"
+/// // stored in buffer 3
+/// // at offset 42
+/// let prefix = "Rust";
+/// let view = ByteView::new(20, prefix.as_bytes())
+///   .with_buffer_index(3)
+///   .with_offset(42);
+///
+/// // create the final u128
+/// let v = view.as_u128();
+/// assert_eq!(v, 0x2a000000037473755200000014);
+/// ```
+///
+/// # Example: decode a `u128` into its constituent fields
+/// ```rust
+/// # use arrow_data::ByteView;
+/// // Convert a u128 to a ByteView
+/// // See validate_{string,binary}_view functions to validate
+/// let v = ByteView::from(0x2a000000037473755200000014);
+///
+/// assert_eq!(v.length, 20);
+/// assert_eq!(v.prefix, 0x74737552);
+/// assert_eq!(v.buffer_index, 3);
+/// assert_eq!(v.offset, 42);
+/// ```
 ///
 /// [`GenericByteViewArray`]: https://docs.rs/arrow/latest/arrow/array/struct.GenericByteViewArray.html
 #[derive(Debug, Copy, Clone, Default)]
@@ -39,6 +71,42 @@ pub struct ByteView {
 }
 
 impl ByteView {
+    /// Construct a [`ByteView`] for data `length` of bytes with the specified prefix.
+    ///
+    /// See example on [`ByteView`] docs
+    ///
+    /// Notes:
+    /// * the length should always be greater than 12 (Data less than 12
+    ///   bytes is stored as an inline view)
+    /// * buffer and offset are set to `0`
+    ///
+    /// # Panics
+    /// If the prefix is not exactly 4 bytes
+    #[inline]
+    pub fn new(length: u32, prefix: &[u8]) -> Self {
+        debug_assert!(length > 12);
+        Self {
+            length,
+            prefix: u32::from_le_bytes(prefix.try_into().unwrap()),
+            buffer_index: 0,
+            offset: 0,
+        }
+    }
+
+    /// Set the [`Self::buffer_index`] field
+    #[inline]
+    pub fn with_buffer_index(mut self, buffer_index: u32) -> Self {
+        self.buffer_index = buffer_index;
+        self
+    }
+
+    /// Set the [`Self::offset`] field
+    #[inline]
+    pub fn with_offset(mut self, offset: u32) -> Self {
+        self.offset = offset;
+        self
+    }
+
     #[inline(always)]
     /// Convert `ByteView` to `u128` by concatenating the fields
     pub fn as_u128(self) -> u128 {
diff --git a/arrow-flight/examples/flight_sql_server.rs b/arrow-flight/examples/flight_sql_server.rs
index 81afecf85..dd3a3943d 100644
--- a/arrow-flight/examples/flight_sql_server.rs
+++ b/arrow-flight/examples/flight_sql_server.rs
@@ -19,6 +19,7 @@ use arrow_flight::sql::server::PeekableFlightDataStream;
 use arrow_flight::sql::DoPutPreparedStatementResult;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
+use core::str;
 use futures::{stream, Stream, TryStreamExt};
 use once_cell::sync::Lazy;
 use prost::Message;
@@ -168,7 +169,7 @@ impl FlightSqlService for FlightSqlServiceImpl {
         let bytes = BASE64_STANDARD
             .decode(base64)
             .map_err(|e| status!("authorization not decodable", e))?;
-        let str = String::from_utf8(bytes).map_err(|e| status!("authorization not parsable", e))?;
+        let str = str::from_utf8(&bytes).map_err(|e| status!("authorization not parsable", e))?;
         let parts: Vec<_> = str.split(':').collect();
         let (user, pass) = match parts.as_slice() {
             [user, pass] => (user, pass),
diff --git a/arrow-flight/src/bin/flight_sql_client.rs b/arrow-flight/src/bin/flight_sql_client.rs
index c334b95a9..8f0618f49 100644
--- a/arrow-flight/src/bin/flight_sql_client.rs
+++ b/arrow-flight/src/bin/flight_sql_client.rs
@@ -26,6 +26,7 @@ use arrow_flight::{
 };
 use arrow_schema::Schema;
 use clap::{Parser, Subcommand};
+use core::str;
 use futures::TryStreamExt;
 use tonic::{
     metadata::MetadataMap,
diff --git a/arrow-flight/src/decode.rs b/arrow-flight/src/decode.rs
index 5561f256c..7bafc3843 100644
--- a/arrow-flight/src/decode.rs
+++ b/arrow-flight/src/decode.rs
@@ -388,11 +388,14 @@ struct FlightStreamState {
 /// FlightData and the decoded payload (Schema, RecordBatch), if any
 #[derive(Debug)]
 pub struct DecodedFlightData {
+    /// The original FlightData message
     pub inner: FlightData,
+    /// The decoded payload
     pub payload: DecodedPayload,
 }
 
 impl DecodedFlightData {
+    /// Create a new DecodedFlightData with no payload
     pub fn new_none(inner: FlightData) -> Self {
         Self {
             inner,
@@ -400,6 +403,7 @@ impl DecodedFlightData {
         }
     }
 
+    /// Create a new DecodedFlightData with a [`Schema`] payload
     pub fn new_schema(inner: FlightData, schema: SchemaRef) -> Self {
         Self {
             inner,
@@ -407,6 +411,7 @@ impl DecodedFlightData {
         }
     }
 
+    /// Create a new [`DecodedFlightData`] with a [`RecordBatch`] payload
     pub fn new_record_batch(inner: FlightData, batch: RecordBatch) -> Self {
         Self {
             inner,
@@ -414,7 +419,7 @@ impl DecodedFlightData {
         }
     }
 
-    /// return the metadata field of the inner flight data
+    /// Return the metadata field of the inner flight data
     pub fn app_metadata(&self) -> Bytes {
         self.inner.app_metadata.clone()
     }
diff --git a/arrow-flight/src/encode.rs b/arrow-flight/src/encode.rs
index 59fa8afd5..55bc92403 100644
--- a/arrow-flight/src/encode.rs
+++ b/arrow-flight/src/encode.rs
@@ -144,6 +144,7 @@ impl Default for FlightDataEncoderBuilder {
 }
 
 impl FlightDataEncoderBuilder {
+    /// Create a new [`FlightDataEncoderBuilder`].
     pub fn new() -> Self {
         Self::default()
     }
@@ -1403,7 +1404,7 @@ mod tests {
         let input_rows = batch.num_rows();
 
         let split = split_batch_for_grpc_response(batch.clone(), max_flight_data_size_bytes);
-        let sizes: Vec<_> = split.iter().map(|batch| batch.num_rows()).collect();
+        let sizes: Vec<_> = split.iter().map(RecordBatch::num_rows).collect();
         let output_rows: usize = sizes.iter().sum();
 
         assert_eq!(sizes, expected_sizes, "mismatch for {batch:?}");
diff --git a/arrow-flight/src/error.rs b/arrow-flight/src/error.rs
index ba979ca9f..499706e1e 100644
--- a/arrow-flight/src/error.rs
+++ b/arrow-flight/src/error.rs
@@ -37,6 +37,7 @@ pub enum FlightError {
 }
 
 impl FlightError {
+    /// Generate a new `FlightError::ProtocolError` variant.
     pub fn protocol(message: impl Into<String>) -> Self {
         Self::ProtocolError(message.into())
     }
@@ -98,6 +99,7 @@ impl From<FlightError> for tonic::Status {
     }
 }
 
+/// Result type for the Apache Arrow Flight crate
 pub type Result<T> = std::result::Result<T, FlightError>;
 
 #[cfg(test)]
diff --git a/arrow-flight/src/lib.rs b/arrow-flight/src/lib.rs
index 64e3ba01c..9f18416c0 100644
--- a/arrow-flight/src/lib.rs
+++ b/arrow-flight/src/lib.rs
@@ -37,6 +37,7 @@
 //!
 //! [Flight SQL]: https://arrow.apache.org/docs/format/FlightSql.html
 #![allow(rustdoc::invalid_html_tags)]
+#![warn(missing_docs)]
 
 use arrow_ipc::{convert, writer, writer::EncodedData, writer::IpcWriteOptions};
 use arrow_schema::{ArrowError, Schema};
@@ -52,6 +53,8 @@ type ArrowResult<T> = std::result::Result<T, ArrowError>;
 
 #[allow(clippy::all)]
 mod gen {
+    // Since this file is auto-generated, we suppress all warnings
+    #![allow(missing_docs)]
     include!("arrow.flight.protocol.rs");
 }
 
@@ -125,6 +128,7 @@ use flight_descriptor::DescriptorType;
 
 /// SchemaAsIpc represents a pairing of a `Schema` with IpcWriteOptions
 pub struct SchemaAsIpc<'a> {
+    /// Data type representing a schema and its IPC write options
     pub pair: (&'a Schema, &'a IpcWriteOptions),
 }
 
@@ -684,6 +688,7 @@ impl PollInfo {
 }
 
 impl<'a> SchemaAsIpc<'a> {
+    /// Create a new `SchemaAsIpc` from a `Schema` and `IpcWriteOptions`
     pub fn new(schema: &'a Schema, options: &'a IpcWriteOptions) -> Self {
         SchemaAsIpc {
             pair: (schema, options),
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index ef52aa27e..e45e505b2 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -695,9 +695,11 @@ fn flight_error_to_arrow_error(err: FlightError) -> ArrowError {
     }
 }
 
-// A polymorphic structure to natively represent different types of data contained in `FlightData`
+/// A polymorphic structure to natively represent different types of data contained in `FlightData`
 pub enum ArrowFlightData {
+    /// A record batch
     RecordBatch(RecordBatch),
+    /// A schema
     Schema(Schema),
 }
 
diff --git a/arrow-flight/src/sql/metadata/sql_info.rs b/arrow-flight/src/sql/metadata/sql_info.rs
index 97304d3c8..2ea30df7f 100644
--- a/arrow-flight/src/sql/metadata/sql_info.rs
+++ b/arrow-flight/src/sql/metadata/sql_info.rs
@@ -331,7 +331,7 @@ impl SqlInfoUnionBuilder {
 ///
 /// Servers constuct - usually static - [`SqlInfoData`] via the [`SqlInfoDataBuilder`],
 /// and build responses using [`CommandGetSqlInfo::into_builder`]
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Default)]
 pub struct SqlInfoDataBuilder {
     /// Use BTreeMap to ensure the values are sorted by value as
     /// to make output consistent
@@ -341,17 +341,10 @@ pub struct SqlInfoDataBuilder {
     infos: BTreeMap<u32, SqlInfoValue>,
 }
 
-impl Default for SqlInfoDataBuilder {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 impl SqlInfoDataBuilder {
+    /// Create a new SQL info builder
     pub fn new() -> Self {
-        Self {
-            infos: BTreeMap::new(),
-        }
+        Self::default()
     }
 
     /// register the specific sql metadata item
diff --git a/arrow-flight/src/sql/metadata/xdbc_info.rs b/arrow-flight/src/sql/metadata/xdbc_info.rs
index 2e635d303..485bedaeb 100644
--- a/arrow-flight/src/sql/metadata/xdbc_info.rs
+++ b/arrow-flight/src/sql/metadata/xdbc_info.rs
@@ -41,24 +41,43 @@ use crate::sql::{CommandGetXdbcTypeInfo, Nullable, Searchable, XdbcDataType, Xdb
 /// Data structure representing type information for xdbc types.
 #[derive(Debug, Clone, Default)]
 pub struct XdbcTypeInfo {
+    /// The name of the type
     pub type_name: String,
+    /// The data type of the type
     pub data_type: XdbcDataType,
+    /// The column size of the type
     pub column_size: Option<i32>,
+    /// The prefix of the type
     pub literal_prefix: Option<String>,
+    /// The suffix of the type
     pub literal_suffix: Option<String>,
+    /// The create parameters of the type
     pub create_params: Option<Vec<String>>,
+    /// The nullability of the type
     pub nullable: Nullable,
+    /// Whether the type is case sensitive
     pub case_sensitive: bool,
+    /// Whether the type is searchable
     pub searchable: Searchable,
+    /// Whether the type is unsigned
     pub unsigned_attribute: Option<bool>,
+    /// Whether the type has fixed precision and scale
     pub fixed_prec_scale: bool,
+    /// Whether the type is auto-incrementing
     pub auto_increment: Option<bool>,
+    /// The local type name of the type
     pub local_type_name: Option<String>,
+    /// The minimum scale of the type
     pub minimum_scale: Option<i32>,
+    /// The maximum scale of the type
     pub maximum_scale: Option<i32>,
+    /// The SQL data type of the type
     pub sql_data_type: XdbcDataType,
+    /// The optional datetime subcode of the type
     pub datetime_subcode: Option<XdbcDatetimeSubcode>,
+    /// The number precision radix of the type
     pub num_prec_radix: Option<i32>,
+    /// The interval precision of the type
     pub interval_precision: Option<i32>,
 }
 
@@ -93,16 +112,6 @@ impl XdbcTypeInfoData {
     }
 }
 
-pub struct XdbcTypeInfoDataBuilder {
-    infos: Vec<XdbcTypeInfo>,
-}
-
-impl Default for XdbcTypeInfoDataBuilder {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 /// A builder for [`XdbcTypeInfoData`] which is used to create [`CommandGetXdbcTypeInfo`] responses.
 ///
 /// # Example
@@ -138,6 +147,16 @@ impl Default for XdbcTypeInfoDataBuilder {
 /// // to access the underlying record batch
 /// let batch = info_list.record_batch(None);
 /// ```
+pub struct XdbcTypeInfoDataBuilder {
+    infos: Vec<XdbcTypeInfo>,
+}
+
+impl Default for XdbcTypeInfoDataBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl XdbcTypeInfoDataBuilder {
     /// Create a new instance of [`XdbcTypeInfoDataBuilder`].
     pub fn new() -> Self {
diff --git a/arrow-flight/src/sql/mod.rs b/arrow-flight/src/sql/mod.rs
index 453f608d3..94bb96a4f 100644
--- a/arrow-flight/src/sql/mod.rs
+++ b/arrow-flight/src/sql/mod.rs
@@ -43,9 +43,11 @@ use bytes::Bytes;
 use paste::paste;
 use prost::Message;
 
+#[allow(clippy::all)]
 mod gen {
-    #![allow(clippy::all)]
     #![allow(rustdoc::unportable_markdown)]
+    // Since this file is auto-generated, we suppress all warnings
+    #![allow(missing_docs)]
     include!("arrow.flight.protocol.sql.rs");
 }
 
@@ -163,7 +165,9 @@ macro_rules! prost_message_ext {
                 /// ```
                 #[derive(Clone, Debug, PartialEq)]
                 pub enum Command {
-                    $($name($name),)*
+                    $(
+                        #[doc = concat!(stringify!($name), "variant")]
+                        $name($name),)*
 
                     /// Any message that is not any FlightSQL command.
                     Unknown(Any),
@@ -297,10 +301,12 @@ pub struct Any {
 }
 
 impl Any {
+    /// Checks whether the message is of type `M`
     pub fn is<M: ProstMessageExt>(&self) -> bool {
         M::type_url() == self.type_url
     }
 
+    /// Unpacks the contents of the message if it is of type `M`
     pub fn unpack<M: ProstMessageExt>(&self) -> Result<Option<M>, ArrowError> {
         if !self.is::<M>() {
             return Ok(None);
@@ -310,6 +316,7 @@ impl Any {
         Ok(Some(m))
     }
 
+    /// Packs a message into an [`Any`] message
     pub fn pack<M: ProstMessageExt>(message: &M) -> Result<Any, ArrowError> {
         Ok(message.as_any())
     }
diff --git a/arrow-flight/src/utils.rs b/arrow-flight/src/utils.rs
index 37d7ff9e7..f6129ddfe 100644
--- a/arrow-flight/src/utils.rs
+++ b/arrow-flight/src/utils.rs
@@ -160,9 +160,12 @@ pub fn batches_to_flight_data(
         dictionaries.extend(encoded_dictionaries.into_iter().map(Into::into));
         flight_data.push(encoded_batch.into());
     }
-    let mut stream = vec![schema_flight_data];
+
+    let mut stream = Vec::with_capacity(1 + dictionaries.len() + flight_data.len());
+
+    stream.push(schema_flight_data);
     stream.extend(dictionaries);
     stream.extend(flight_data);
-    let flight_data: Vec<_> = stream.into_iter().collect();
+    let flight_data = stream;
     Ok(flight_data)
 }
diff --git a/arrow-ipc/src/convert.rs b/arrow-ipc/src/convert.rs
index 52c6a0d61..eef236529 100644
--- a/arrow-ipc/src/convert.rs
+++ b/arrow-ipc/src/convert.rs
@@ -133,6 +133,7 @@ pub fn schema_to_fb(schema: &Schema) -> FlatBufferBuilder<'_> {
     IpcSchemaEncoder::new().schema_to_fb(schema)
 }
 
+/// Push a key-value metadata into a FlatBufferBuilder and return [WIPOffset]
 pub fn metadata_to_fb<'a>(
     fbb: &mut FlatBufferBuilder<'a>,
     metadata: &HashMap<String, String>,
@@ -152,7 +153,7 @@ pub fn metadata_to_fb<'a>(
     fbb.create_vector(&custom_metadata)
 }
 
-#[deprecated(since = "54.0.0", note = "Use `IpcSchemaConverter`.")]
+/// Adds a [Schema] to a flatbuffer and returns the offset
 pub fn schema_to_fb_offset<'a>(
     fbb: &mut FlatBufferBuilder<'a>,
     schema: &Schema,
diff --git a/arrow-ipc/src/lib.rs b/arrow-ipc/src/lib.rs
index 4f35ffb60..dde137153 100644
--- a/arrow-ipc/src/lib.rs
+++ b/arrow-ipc/src/lib.rs
@@ -19,6 +19,7 @@
 //!
 //! [Arrow IPC Format]: https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
 
+#![warn(missing_docs)]
 pub mod convert;
 pub mod reader;
 pub mod writer;
@@ -31,6 +32,7 @@ mod compression;
 #[allow(clippy::redundant_static_lifetimes)]
 #[allow(clippy::redundant_field_names)]
 #[allow(non_camel_case_types)]
+#[allow(missing_docs)] // Because this is autogenerated
 pub mod gen;
 
 pub use self::gen::File::*;
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index b5cf20ef3..f9256b4e8 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -60,7 +60,7 @@ pub struct IpcWriteOptions {
     /// Compression, if desired. Will result in a runtime error
     /// if the corresponding feature is not enabled
     batch_compression_type: Option<crate::CompressionType>,
-    /// Flag indicating whether the writer should preserver the dictionary IDs defined in the
+    /// Flag indicating whether the writer should preserve the dictionary IDs defined in the
     /// schema or generate unique dictionary IDs internally during encoding.
     ///
     /// Defaults to `true`
@@ -135,6 +135,8 @@ impl IpcWriteOptions {
         }
     }
 
+    /// Return whether the writer is configured to preserve the dictionary IDs
+    /// defined in the schema
     pub fn preserve_dict_id(&self) -> bool {
         self.preserve_dict_id
     }
@@ -200,6 +202,11 @@ impl Default for IpcWriteOptions {
 pub struct IpcDataGenerator {}
 
 impl IpcDataGenerator {
+    /// Converts a schema to an IPC message along with `dictionary_tracker`
+    /// and returns it encoded inside [EncodedData] as a flatbuffer
+    ///
+    /// Preferred method over [IpcDataGenerator::schema_to_bytes] since it's
+    /// deprecated since Arrow v54.0.0
     pub fn schema_to_bytes_with_dictionary_tracker(
         &self,
         schema: &Schema,
@@ -234,6 +241,7 @@ impl IpcDataGenerator {
         since = "54.0.0",
         note = "Use `schema_to_bytes_with_dictionary_tracker` instead. This function signature of `schema_to_bytes_with_dictionary_tracker` in the next release."
     )]
+    /// Converts a schema to an IPC message and returns it encoded inside [EncodedData] as a flatbuffer
     pub fn schema_to_bytes(&self, schema: &Schema, write_options: &IpcWriteOptions) -> EncodedData {
         let mut fbb = FlatBufferBuilder::new();
         let schema = {
@@ -951,6 +959,7 @@ impl<W: Write> FileWriter<W> {
         })
     }
 
+    /// Adds a key-value pair to the [FileWriter]'s custom metadata
     pub fn write_metadata(&mut self, key: impl Into<String>, value: impl Into<String>) {
         self.custom_metadata.insert(key.into(), value.into());
     }
diff --git a/arrow-json/src/writer.rs b/arrow-json/src/writer.rs
index 86d2e88d9..d973206cc 100644
--- a/arrow-json/src/writer.rs
+++ b/arrow-json/src/writer.rs
@@ -397,6 +397,7 @@ where
 
 #[cfg(test)]
 mod tests {
+    use core::str;
     use std::fs::{read_to_string, File};
     use std::io::{BufReader, Seek};
     use std::sync::Arc;
@@ -1111,7 +1112,7 @@ mod tests {
             }
         }
 
-        let result = String::from_utf8(buf).unwrap();
+        let result = str::from_utf8(&buf).unwrap();
         let expected = read_to_string(test_file).unwrap();
         for (r, e) in result.lines().zip(expected.lines()) {
             let mut expected_json = serde_json::from_str::<Value>(e).unwrap();
@@ -1150,7 +1151,7 @@ mod tests {
     fn json_writer_empty() {
         let mut writer = ArrayWriter::new(vec![] as Vec<u8>);
         writer.finish().unwrap();
-        assert_eq!(String::from_utf8(writer.into_inner()).unwrap(), "");
+        assert_eq!(str::from_utf8(&writer.into_inner()).unwrap(), "");
     }
 
     #[test]
@@ -1279,7 +1280,7 @@ mod tests {
             writer.write(&batch).unwrap();
         }
 
-        let result = String::from_utf8(buf).unwrap();
+        let result = str::from_utf8(&buf).unwrap();
         let expected = read_to_string(test_file).unwrap();
         for (r, e) in result.lines().zip(expected.lines()) {
             let mut expected_json = serde_json::from_str::<Value>(e).unwrap();
@@ -1321,7 +1322,7 @@ mod tests {
             writer.write_batches(&batches).unwrap();
         }
 
-        let result = String::from_utf8(buf).unwrap();
+        let result = str::from_utf8(&buf).unwrap();
         let expected = read_to_string(test_file).unwrap();
         // result is eq to 2 same batches
         let expected = format!("{expected}\n{expected}");
diff --git a/arrow-schema/src/field.rs b/arrow-schema/src/field.rs
index fc4852a3d..b532ea861 100644
--- a/arrow-schema/src/field.rs
+++ b/arrow-schema/src/field.rs
@@ -610,14 +610,14 @@ mod test {
     #[test]
     fn test_new_with_string() {
         // Fields should allow owned Strings to support reuse
-        let s = String::from("c1");
+        let s = "c1";
         Field::new(s, DataType::Int64, false);
     }
 
     #[test]
     fn test_new_dict_with_string() {
         // Fields should allow owned Strings to support reuse
-        let s = String::from("c1");
+        let s = "c1";
         Field::new_dict(s, DataType::Int64, false, 4, false);
     }
 
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index 75acb73e5..c52c3f8df 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -44,7 +44,6 @@ static DEFAULT_METADATA_ENDPOINT: &str = "http://169.254.169.254";
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 enum Error {
     #[snafu(display("Missing bucket name"))]
     MissingBucketName,
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 6fe4889db..7034a372e 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -65,7 +65,6 @@ const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-amz-meta-";
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 pub(crate) enum Error {
     #[snafu(display("Error performing DeleteObjects request: {}", source))]
     DeleteObjectsRequest { source: crate::client::retry::Error },
diff --git a/object_store/src/aws/resolve.rs b/object_store/src/aws/resolve.rs
index 12c9f26d2..4c7489316 100644
--- a/object_store/src/aws/resolve.rs
+++ b/object_store/src/aws/resolve.rs
@@ -21,7 +21,6 @@ use snafu::{ensure, OptionExt, ResultExt, Snafu};
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 enum Error {
     #[snafu(display("Bucket '{}' not found", bucket))]
     BucketNotFound { bucket: String },
diff --git a/object_store/src/azure/builder.rs b/object_store/src/azure/builder.rs
index 35cedeafc..1c4589ba1 100644
--- a/object_store/src/azure/builder.rs
+++ b/object_store/src/azure/builder.rs
@@ -46,7 +46,6 @@ const MSI_ENDPOINT_ENV_KEY: &str = "IDENTITY_ENDPOINT";
 
 /// A specialized `Error` for Azure builder-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 enum Error {
     #[snafu(display("Unable parse source url. Url: {}, Error: {}", url, source))]
     UnableToParseUrl {
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index 049905155..06d3fb5c8 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -60,7 +60,6 @@ static TAGS_HEADER: HeaderName = HeaderName::from_static("x-ms-tags");
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 pub(crate) enum Error {
     #[snafu(display("Error performing get request {}: {}", path, source))]
     GetRequest {
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 0fef5785c..ae6a8d9de 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -96,7 +96,6 @@ impl ContentRange {
 
 /// A specialized `Error` for get-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 enum GetResultError {
     #[snafu(context(false))]
     Header {
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 8820983b2..a0d83eb0b 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -1224,78 +1224,116 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 #[non_exhaustive]
 pub enum Error {
+    /// A fallback error type when no variant matches
     #[snafu(display("Generic {} error: {}", store, source))]
     Generic {
+        /// The store this error originated from
         store: &'static str,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when the object is not found at given location
     #[snafu(display("Object at location {} not found: {}", path, source))]
     NotFound {
+        /// The path to file
         path: String,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error for invalid path
     #[snafu(
         display("Encountered object with invalid path: {}", source),
         context(false)
     )]
-    InvalidPath { source: path::Error },
+    InvalidPath {
+        /// The wrapped error
+        source: path::Error,
+    },
 
+    /// Error when `tokio::spawn` failed
     #[snafu(display("Error joining spawned task: {}", source), context(false))]
-    JoinError { source: tokio::task::JoinError },
+    JoinError {
+        /// The wrapped error
+        source: tokio::task::JoinError,
+    },
 
+    /// Error when the attempted operation is not supported
     #[snafu(display("Operation not supported: {}", source))]
     NotSupported {
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when the object already exists
     #[snafu(display("Object at location {} already exists: {}", path, source))]
     AlreadyExists {
+        /// The path to the
         path: String,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when the required conditions failed for the operation
     #[snafu(display("Request precondition failure for path {}: {}", path, source))]
     Precondition {
+        /// The path to the file
         path: String,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when the object at the location isn't modified
     #[snafu(display("Object at location {} not modified: {}", path, source))]
     NotModified {
+        /// The path to the file
         path: String,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when an operation is not implemented
     #[snafu(display("Operation not yet implemented."))]
     NotImplemented,
 
+    /// Error when the used credentials don't have enough permission
+    /// to perform the requested operation
     #[snafu(display(
         "The operation lacked the necessary privileges to complete for path {}: {}",
         path,
         source
     ))]
     PermissionDenied {
+        /// The path to the file
         path: String,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when the used credentials lack valid authentication
     #[snafu(display(
         "The operation lacked valid authentication credentials for path {}: {}",
         path,
         source
     ))]
     Unauthenticated {
+        /// The path to the file
         path: String,
+        /// The wrapped error
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
+    /// Error when a configuration key is invalid for the store used
     #[snafu(display("Configuration key: '{}' is not valid for store '{}'.", key, store))]
-    UnknownConfigurationKey { store: &'static str, key: String },
+    UnknownConfigurationKey {
+        /// The object store used
+        store: &'static str,
+        /// The configuration key used
+        key: String,
+    },
 }
 
 impl From<Error> for std::io::Error {
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index db4b4b050..ac10f332d 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -44,7 +44,6 @@ use crate::{
 
 /// A specialized `Error` for filesystem object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 pub(crate) enum Error {
     #[snafu(display("File size for {} did not fit in a usize: {}", path, source))]
     FileSizeOverflowedUsize {
diff --git a/object_store/src/memory.rs b/object_store/src/memory.rs
index 0d72983b0..b458bdddf 100644
--- a/object_store/src/memory.rs
+++ b/object_store/src/memory.rs
@@ -38,7 +38,6 @@ use crate::{GetOptions, PutPayload};
 
 /// A specialized `Error` for in-memory object store-related errors
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 enum Error {
     #[snafu(display("No data in memory found. Location: {path}"))]
     NoDataInMemory { path: String },
diff --git a/object_store/src/path/mod.rs b/object_store/src/path/mod.rs
index 59e08e2ea..4c9bb5f05 100644
--- a/object_store/src/path/mod.rs
+++ b/object_store/src/path/mod.rs
@@ -36,32 +36,57 @@ pub use parts::{InvalidPart, PathPart};
 
 /// Error returned by [`Path::parse`]
 #[derive(Debug, Snafu)]
-#[allow(missing_docs)]
 #[non_exhaustive]
 pub enum Error {
+    /// Error when there's an empty segment between two slashes `/` in the path
     #[snafu(display("Path \"{}\" contained empty path segment", path))]
-    EmptySegment { path: String },
+    EmptySegment {
+        /// The source path
+        path: String,
+    },
 
+    /// Error when an invalid segment is encountered in the given path
     #[snafu(display("Error parsing Path \"{}\": {}", path, source))]
-    BadSegment { path: String, source: InvalidPart },
+    BadSegment {
+        /// The source path
+        path: String,
+        /// The part containing the error
+        source: InvalidPart,
+    },
 
+    /// Error when path cannot be canonicalized
     #[snafu(display("Failed to canonicalize path \"{}\": {}", path.display(), source))]
     Canonicalize {
+        /// The source path
         path: std::path::PathBuf,
+        /// The underlying error
         source: std::io::Error,
     },
 
+    /// Error when the path is not a valid URL
     #[snafu(display("Unable to convert path \"{}\" to URL", path.display()))]
-    InvalidPath { path: std::path::PathBuf },
+    InvalidPath {
+        /// The source path
+        path: std::path::PathBuf,
+    },
 
+    /// Error when a path contains non-unicode characters
     #[snafu(display("Path \"{}\" contained non-unicode characters: {}", path, source))]
     NonUnicode {
+        /// The source path
         path: String,
+        /// The underlying `UTF8Error`
         source: std::str::Utf8Error,
     },
 
+    /// Error when the a path doesn't start with given prefix
     #[snafu(display("Path {} does not start with prefix {}", path, prefix))]
-    PrefixMismatch { path: String, prefix: String },
+    PrefixMismatch {
+        /// The source path
+        path: String,
+        /// The mismatched prefix
+        prefix: String,
+    },
 }
 
 /// A parsed path representation that can be safely written to object storage
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs b/parquet/src/arrow/arrow_reader/statistics.rs
index 4c0dac05c..8a7511be2 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -1159,7 +1159,7 @@ where
 ///
 /// # Schemas
 ///
-/// The converter ues the schema of the Parquet file and the Arrow schema to
+/// The converter uses the schema of the Parquet file and the Arrow schema to
 /// convert the underlying statistics value (stored as a parquet value) into the
 /// corresponding Arrow value. For example, Decimals are stored as binary in
 /// parquet files and this structure handles mapping them to the `i128`
@@ -1175,6 +1175,8 @@ pub struct StatisticsConverter<'a> {
     parquet_column_index: Option<usize>,
     /// The field (with data type) of the column in the Arrow schema
     arrow_field: &'a Field,
+    /// treat missing null_counts as 0 nulls
+    missing_null_counts_as_zero: bool,
 }
 
 impl<'a> StatisticsConverter<'a> {
@@ -1191,6 +1193,23 @@ impl<'a> StatisticsConverter<'a> {
         self.arrow_field
     }
 
+    /// Set the statistics converter to treat missing null counts as missing
+    ///
+    /// By default, the converter will treat missing null counts as though
+    /// the null count is known to be `0`.
+    ///
+    /// Note that parquet files written by parquet-rs currently do not store
+    /// null counts even when it is known there are zero nulls, and the reader
+    /// will return 0 for the null counts in that instance. This behavior may
+    /// change in a future release.
+    ///
+    /// Both parquet-java and parquet-cpp store null counts as 0 when there are
+    /// no nulls, and don't write unknown values to the null count field.
+    pub fn with_missing_null_counts_as_zero(mut self, missing_null_counts_as_zero: bool) -> Self {
+        self.missing_null_counts_as_zero = missing_null_counts_as_zero;
+        self
+    }
+
     /// Returns a [`UInt64Array`] with row counts for each row group
     ///
     /// # Return Value
@@ -1284,6 +1303,7 @@ impl<'a> StatisticsConverter<'a> {
         Ok(Self {
             parquet_column_index: parquet_index,
             arrow_field,
+            missing_null_counts_as_zero: true,
         })
     }
 
@@ -1382,7 +1402,15 @@ impl<'a> StatisticsConverter<'a> {
         let null_counts = metadatas
             .into_iter()
             .map(|x| x.column(parquet_index).statistics())
-            .map(|s| s.and_then(|s| s.null_count_opt()));
+            .map(|s| {
+                s.and_then(|s| {
+                    if self.missing_null_counts_as_zero {
+                        Some(s.null_count_opt().unwrap_or(0))
+                    } else {
+                        s.null_count_opt()
+                    }
+                })
+            });
         Ok(UInt64Array::from_iter(null_counts))
     }
 
@@ -1593,3 +1621,5 @@ impl<'a> StatisticsConverter<'a> {
         new_null_array(data_type, num_row_groups)
     }
 }
+
+// See tests in parquet/tests/arrow_reader/statistics.rs
diff --git a/parquet/src/arrow/async_reader/metadata.rs b/parquet/src/arrow/async_reader/metadata.rs
index cd45d2abd..b7fac6fe7 100644
--- a/parquet/src/arrow/async_reader/metadata.rs
+++ b/parquet/src/arrow/async_reader/metadata.rs
@@ -52,6 +52,7 @@ impl<F: MetadataFetch> MetadataLoader<F> {
     /// Create a new [`MetadataLoader`] by reading the footer information
     ///
     /// See [`fetch_parquet_metadata`] for the meaning of the individual parameters
+    #[deprecated(since = "53.1.0", note = "Use ParquetMetaDataReader")]
     pub async fn load(mut fetch: F, file_size: usize, prefetch: Option<usize>) -> Result<Self> {
         if file_size < FOOTER_SIZE {
             return Err(ParquetError::EOF(format!(
@@ -108,6 +109,7 @@ impl<F: MetadataFetch> MetadataLoader<F> {
     }
 
     /// Create a new [`MetadataLoader`] from an existing [`ParquetMetaData`]
+    #[deprecated(since = "53.1.0", note = "Use ParquetMetaDataReader")]
     pub fn new(fetch: F, metadata: ParquetMetaData) -> Self {
         Self {
             fetch,
@@ -120,6 +122,7 @@ impl<F: MetadataFetch> MetadataLoader<F> {
     ///
     /// * `column_index`: if true will load column index
     /// * `offset_index`: if true will load offset index
+    #[deprecated(since = "53.1.0", note = "Use ParquetMetaDataReader")]
     pub async fn load_page_index(&mut self, column_index: bool, offset_index: bool) -> Result<()> {
         if !column_index && !offset_index {
             return Ok(());
@@ -226,6 +229,7 @@ where
 /// in the first request, instead of 8, and only issue further requests
 /// if additional bytes are needed. Providing a `prefetch` hint can therefore
 /// significantly reduce the number of `fetch` requests, and consequently latency
+#[deprecated(since = "53.1.0", note = "Use ParquetMetaDataReader")]
 pub async fn fetch_parquet_metadata<F, Fut>(
     fetch: F,
     file_size: usize,
@@ -236,10 +240,14 @@ where
     Fut: Future<Output = Result<Bytes>> + Send,
 {
     let fetch = MetadataFetchFn(fetch);
-    let loader = MetadataLoader::load(fetch, file_size, prefetch).await?;
-    Ok(loader.finish())
+    ParquetMetaDataReader::new()
+        .with_prefetch_hint(prefetch)
+        .load_and_finish(fetch, file_size)
+        .await
 }
 
+// these tests are all replicated in parquet::file::metadata::reader
+#[allow(deprecated)]
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
index 89e4d6adb..5e8bdbc02 100644
--- a/parquet/src/arrow/async_reader/mod.rs
+++ b/parquet/src/arrow/async_reader/mod.rs
@@ -212,6 +212,8 @@ impl ArrowReaderMetadata {
         input: &mut T,
         options: ArrowReaderOptions,
     ) -> Result<Self> {
+        // TODO: this is all rather awkward. It would be nice if AsyncFileReader::get_metadata
+        // took an argument to fetch the page indexes.
         let mut metadata = input.get_metadata().await?;
 
         if options.page_index
@@ -219,9 +221,9 @@ impl ArrowReaderMetadata {
             && metadata.offset_index().is_none()
         {
             let m = Arc::try_unwrap(metadata).unwrap_or_else(|e| e.as_ref().clone());
-            let mut loader = MetadataLoader::new(input, m);
-            loader.load_page_index(true, true).await?;
-            metadata = Arc::new(loader.finish())
+            let mut reader = ParquetMetaDataReader::new_with_metadata(m).with_page_indexes(true);
+            reader.load_page_index(input).await?;
+            metadata = Arc::new(reader.finish()?)
         }
         Self::try_new(metadata, options)
     }
diff --git a/parquet/src/arrow/async_reader/store.rs b/parquet/src/arrow/async_reader/store.rs
index 77c00e91a..e6b47856e 100644
--- a/parquet/src/arrow/async_reader/store.rs
+++ b/parquet/src/arrow/async_reader/store.rs
@@ -24,9 +24,9 @@ use futures::{FutureExt, TryFutureExt};
 
 use object_store::{ObjectMeta, ObjectStore};
 
-use crate::arrow::async_reader::{AsyncFileReader, MetadataLoader};
+use crate::arrow::async_reader::AsyncFileReader;
 use crate::errors::Result;
-use crate::file::metadata::ParquetMetaData;
+use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
 
 /// Reads Parquet files in object storage using [`ObjectStore`].
 ///
@@ -124,15 +124,14 @@ impl AsyncFileReader for ParquetObjectReader {
 
     fn get_metadata(&mut self) -> BoxFuture<'_, Result<Arc<ParquetMetaData>>> {
         Box::pin(async move {
-            let preload_column_index = self.preload_column_index;
-            let preload_offset_index = self.preload_offset_index;
             let file_size = self.meta.size;
-            let prefetch = self.metadata_size_hint;
-            let mut loader = MetadataLoader::load(self, file_size, prefetch).await?;
-            loader
-                .load_page_index(preload_column_index, preload_offset_index)
+            let metadata = ParquetMetaDataReader::new()
+                .with_column_indexes(self.preload_column_index)
+                .with_offset_indexes(self.preload_offset_index)
+                .with_prefetch_hint(self.metadata_size_hint)
+                .load_and_finish(self, file_size)
                 .await?;
-            Ok(Arc::new(loader.finish()))
+            Ok(Arc::new(metadata))
         })
     }
 }
diff --git a/parquet/src/compression.rs b/parquet/src/compression.rs
index edf675f13..ccc060250 100644
--- a/parquet/src/compression.rs
+++ b/parquet/src/compression.rs
@@ -298,6 +298,35 @@ mod gzip_codec {
 pub use gzip_codec::*;
 
 /// Represents a valid gzip compression level.
+///
+/// Defaults to 6.
+///
+/// * 0: least compression
+/// * 9: most compression (that other software can read)
+/// * 10: most compression (incompatible with other software, see below)
+/// #### WARNING:
+/// Level 10 compression can offer smallest file size,
+/// but Parquet files created with it will not be readable
+/// by other "standard" paquet readers.
+///
+/// Do **NOT** use level 10 if you need other software to
+/// be able to read the files. Read below for details.
+///
+/// ### IMPORTANT:
+/// There's often confusion about the compression levels in `flate2` vs `arrow`
+/// as highlighted in issue [#1011](https://github.com/apache/arrow-rs/issues/6282).
+///
+/// `flate2` supports two compression backends: `miniz_oxide` and `zlib`.
+///
+/// - `zlib` supports levels from 0 to 9.
+/// - `miniz_oxide` supports levels from 0 to 10.
+///
+/// `arrow` uses `flate` with `rust_backend` feature,
+/// which provides `miniz_oxide` as the backend.
+/// Therefore 0-10 levels are supported.
+///
+/// `flate2` documents this behavior properly with
+/// [this commit](https://github.com/rust-lang/flate2-rs/pull/430).
 #[derive(Debug, Eq, PartialEq, Hash, Clone, Copy)]
 pub struct GzipLevel(u32);
 
diff --git a/parquet/src/file/metadata/reader.rs b/parquet/src/file/metadata/reader.rs
index 9e00c6860..88f53d02f 100644
--- a/parquet/src/file/metadata/reader.rs
+++ b/parquet/src/file/metadata/reader.rs
@@ -329,13 +329,18 @@ impl ParquetMetaDataReader {
             return Ok(());
         }
 
-        self.load_page_index(fetch, remainder).await
+        self.load_page_index_with_remainder(fetch, remainder).await
     }
 
     /// Asynchronously fetch the page index structures when a [`ParquetMetaData`] has already
     /// been obtained. See [`Self::new_with_metadata()`].
     #[cfg(feature = "async")]
-    pub async fn load_page_index<F: MetadataFetch>(
+    pub async fn load_page_index<F: MetadataFetch>(&mut self, fetch: F) -> Result<()> {
+        self.load_page_index_with_remainder(fetch, None).await
+    }
+
+    #[cfg(feature = "async")]
+    async fn load_page_index_with_remainder<F: MetadataFetch>(
         &mut self,
         mut fetch: F,
         remainder: Option<(usize, Bytes)>,
diff --git a/parquet/src/file/metadata/writer.rs b/parquet/src/file/metadata/writer.rs
index db78606e4..44328c635 100644
--- a/parquet/src/file/metadata/writer.rs
+++ b/parquet/src/file/metadata/writer.rs
@@ -516,7 +516,7 @@ mod tests {
     /// Temporary function so we can test loading metadata with page indexes
     /// while we haven't fully figured out how to load it cleanly
     async fn load_metadata_from_bytes(file_size: usize, data: Bytes) -> ParquetMetaData {
-        use crate::arrow::async_reader::{MetadataFetch, MetadataLoader};
+        use crate::arrow::async_reader::MetadataFetch;
         use crate::errors::Result as ParquetResult;
         use futures::future::BoxFuture;
         use futures::FutureExt;
@@ -569,13 +569,11 @@ mod tests {
             Box::new(AsyncBytes::new(data)),
             file_size - metadata_length..file_size,
         );
-        let metadata = MetadataLoader::load(&mut reader, file_size, None)
+        ParquetMetaDataReader::new()
+            .with_page_indexes(true)
+            .load_and_finish(&mut reader, file_size)
             .await
-            .unwrap();
-        let loaded_metadata = metadata.finish();
-        let mut metadata = MetadataLoader::new(&mut reader, loaded_metadata);
-        metadata.load_page_index(true, true).await.unwrap();
-        metadata.finish()
+            .unwrap()
     }
 
     fn check_columns_are_equivalent(left: &ColumnChunkMetaData, right: &ColumnChunkMetaData) {
diff --git a/parquet/src/file/reader.rs b/parquet/src/file/reader.rs
index cff921b20..d8a61fafe 100644
--- a/parquet/src/file/reader.rs
+++ b/parquet/src/file/reader.rs
@@ -45,25 +45,38 @@ pub trait Length {
     fn len(&self) -> u64;
 }
 
-/// The ChunkReader trait generates readers of chunks of a source.
+/// Generates [`Read`]ers to read chunks of a Parquet data source.
 ///
-/// For more information see [`File::try_clone`]
+/// The Parquet reader uses [`ChunkReader`] to access Parquet data, allowing
+/// multiple decoders to read concurrently from different locations in the same file.
+///
+/// The trait provides:
+/// * random access (via [`Self::get_bytes`])
+/// * sequential (via [`Self::get_read`])
+///
+/// # Provided Implementations
+/// * [`File`] for reading from local file system
+/// * [`Bytes`] for reading from an in-memory buffer
+///
+/// User provided implementations can implement more sophisticated behaviors
+/// such as on-demand buffering or scan sharing.
 pub trait ChunkReader: Length + Send + Sync {
+    /// The concrete type of readers returned by this trait
     type T: Read;
 
-    /// Get a [`Read`] starting at the provided file offset
+    /// Get a [`Read`] instance starting at the provided file offset
     ///
-    /// Subsequent or concurrent calls to [`Self::get_read`] or [`Self::get_bytes`] may
-    /// side-effect on previously returned [`Self::T`]. Care should be taken to avoid this
-    ///
-    /// See [`File::try_clone`] for more information
+    /// Returned readers follow the model of [`File::try_clone`] where mutations
+    /// of one reader affect all readers. Thus subsequent or concurrent calls to
+    /// [`Self::get_read`] or [`Self::get_bytes`] may cause side-effects on
+    /// previously returned readers. Callers of `get_read` should take care
+    /// to avoid race conditions.
     fn get_read(&self, start: u64) -> Result<Self::T>;
 
-    /// Get a range as bytes
-    ///
-    /// Concurrent calls to [`Self::get_bytes`] may result in interleaved output
+    /// Get a range of data in memory as [`Bytes`]
     ///
-    /// See [`File::try_clone`] for more information
+    /// Similarly to [`Self::get_read`], this method may have side-effects on
+    /// previously returned readers.
     fn get_bytes(&self, start: u64, length: usize) -> Result<Bytes>;
 }
 


Test patch:
diff --git a/arrow-integration-test/src/lib.rs b/arrow-integration-test/src/lib.rs
index d1486fd5a..ea5b545f2 100644
--- a/arrow-integration-test/src/lib.rs
+++ b/arrow-integration-test/src/lib.rs
@@ -21,6 +21,7 @@
 //!
 //! This is not a canonical format, but provides a human-readable way of verifying language implementations
 
+#![warn(missing_docs)]
 use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano, ScalarBuffer};
 use hex::decode;
 use num::BigInt;
@@ -49,8 +50,11 @@ pub use schema::*;
 /// See <https://github.com/apache/arrow/blob/master/docs/source/format/Integration.rst#json-test-data-format>
 #[derive(Deserialize, Serialize, Debug)]
 pub struct ArrowJson {
+    /// The Arrow schema for JSON file
     pub schema: ArrowJsonSchema,
+    /// The `RecordBatch`es in the JSON file
     pub batches: Vec<ArrowJsonBatch>,
+    /// The dictionaries in the JSON file
     #[serde(skip_serializing_if = "Option::is_none")]
     pub dictionaries: Option<Vec<ArrowJsonDictionaryBatch>>,
 }
@@ -60,7 +64,9 @@ pub struct ArrowJson {
 /// Fields are left as JSON `Value` as they vary by `DataType`
 #[derive(Deserialize, Serialize, Debug)]
 pub struct ArrowJsonSchema {
+    /// An array of JSON fields
     pub fields: Vec<ArrowJsonField>,
+    /// An array of metadata key-value pairs
     #[serde(skip_serializing_if = "Option::is_none")]
     pub metadata: Option<Vec<HashMap<String, String>>>,
 }
@@ -68,13 +74,20 @@ pub struct ArrowJsonSchema {
 /// Fields are left as JSON `Value` as they vary by `DataType`
 #[derive(Deserialize, Serialize, Debug)]
 pub struct ArrowJsonField {
+    /// The name of the field
     pub name: String,
+    /// The data type of the field,
+    /// can be any valid JSON value
     #[serde(rename = "type")]
     pub field_type: Value,
+    /// Whether the field is nullable
     pub nullable: bool,
+    /// The children fields
     pub children: Vec<ArrowJsonField>,
+    /// The dictionary for the field
     #[serde(skip_serializing_if = "Option::is_none")]
     pub dictionary: Option<ArrowJsonFieldDictionary>,
+    /// The metadata for the field, if any
     #[serde(skip_serializing_if = "Option::is_none")]
     pub metadata: Option<Value>,
 }
@@ -115,20 +128,28 @@ impl From<&Field> for ArrowJsonField {
     }
 }
 
+/// Represents a dictionary-encoded field in the Arrow JSON format
 #[derive(Deserialize, Serialize, Debug)]
 pub struct ArrowJsonFieldDictionary {
+    /// A unique identifier for the dictionary
     pub id: i64,
+    /// The type of the dictionary index
     #[serde(rename = "indexType")]
     pub index_type: DictionaryIndexType,
+    /// Whether the dictionary is ordered
     #[serde(rename = "isOrdered")]
     pub is_ordered: bool,
 }
 
+/// Type of an index for a dictionary-encoded field in the Arrow JSON format
 #[derive(Deserialize, Serialize, Debug)]
 pub struct DictionaryIndexType {
+    /// The name of the dictionary index type
     pub name: String,
+    /// Whether the dictionary index type is signed
     #[serde(rename = "isSigned")]
     pub is_signed: bool,
+    /// The bit width of the dictionary index type
     #[serde(rename = "bitWidth")]
     pub bit_width: i64,
 }
@@ -137,6 +158,7 @@ pub struct DictionaryIndexType {
 #[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct ArrowJsonBatch {
     count: usize,
+    /// The columns in the record batch
     pub columns: Vec<ArrowJsonColumn>,
 }
 
@@ -144,7 +166,9 @@ pub struct ArrowJsonBatch {
 #[derive(Deserialize, Serialize, Debug, Clone)]
 #[allow(non_snake_case)]
 pub struct ArrowJsonDictionaryBatch {
+    /// The unique identifier for the dictionary
     pub id: i64,
+    /// The data for the dictionary
     pub data: ArrowJsonBatch,
 }
 
@@ -152,15 +176,21 @@ pub struct ArrowJsonDictionaryBatch {
 #[derive(Deserialize, Serialize, Clone, Debug)]
 pub struct ArrowJsonColumn {
     name: String,
+    /// The number of elements in the column
     pub count: usize,
+    /// The validity bitmap to determine null values
     #[serde(rename = "VALIDITY")]
     pub validity: Option<Vec<u8>>,
+    /// The data values in the column
     #[serde(rename = "DATA")]
     pub data: Option<Vec<Value>>,
+    /// The offsets for variable-sized data types
     #[serde(rename = "OFFSET")]
     pub offset: Option<Vec<Value>>, // leaving as Value as 64-bit offsets are strings
+    /// The type id for union types
     #[serde(rename = "TYPE_ID")]
     pub type_id: Option<Vec<i8>>,
+    /// The children columns for nested types
     pub children: Option<Vec<ArrowJsonColumn>>,
 }
 
@@ -189,6 +219,7 @@ impl ArrowJson {
         Ok(true)
     }
 
+    /// Convert the stored dictionaries to `Vec[RecordBatch]`
     pub fn get_record_batches(&self) -> Result<Vec<RecordBatch>> {
         let schema = self.schema.to_arrow_schema()?;
 
@@ -275,6 +306,7 @@ impl ArrowJsonField {
     }
 }
 
+/// Generates a [`RecordBatch`] from an Arrow JSON batch, given a schema
 pub fn record_batch_from_json(
     schema: &Schema,
     json_batch: ArrowJsonBatch,
@@ -877,6 +909,7 @@ pub fn array_from_json(
     }
 }
 
+/// Construct a [`DictionaryArray`] from a partially typed JSON column
 pub fn dictionary_array_from_json(
     field: &Field,
     json_col: ArrowJsonColumn,
@@ -965,6 +998,7 @@ fn create_null_buf(json_col: &ArrowJsonColumn) -> Buffer {
 }
 
 impl ArrowJsonBatch {
+    /// Convert a [`RecordBatch`] to an [`ArrowJsonBatch`]
     pub fn from_batch(batch: &RecordBatch) -> ArrowJsonBatch {
         let mut json_batch = ArrowJsonBatch {
             count: batch.num_rows(),
diff --git a/arrow-integration-testing/src/flight_client_scenarios/auth_basic_proto.rs b/arrow-integration-testing/src/flight_client_scenarios/auth_basic_proto.rs
index 376e31e15..34c3c7706 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/auth_basic_proto.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/auth_basic_proto.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Scenario for testing basic auth.
+
 use crate::{AUTH_PASSWORD, AUTH_USERNAME};
 
 use arrow_flight::{flight_service_client::FlightServiceClient, BasicAuth, HandshakeRequest};
@@ -27,6 +29,7 @@ type Result<T = (), E = Error> = std::result::Result<T, E>;
 
 type Client = FlightServiceClient<tonic::transport::Channel>;
 
+/// Run a scenario that tests basic auth.
 pub async fn run_scenario(host: &str, port: u16) -> Result {
     let url = format!("http://{host}:{port}");
     let mut client = FlightServiceClient::connect(url).await?;
diff --git a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
index 1a6c4e28a..c8289ff44 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Integration tests for the Flight client.
+
 use crate::open_json_file;
 use std::collections::HashMap;
 
@@ -40,6 +42,7 @@ type Result<T = (), E = Error> = std::result::Result<T, E>;
 
 type Client = FlightServiceClient<tonic::transport::Channel>;
 
+/// Run a scenario that uploads data to a Flight server and then downloads it back
 pub async fn run_scenario(host: &str, port: u16, path: &str) -> Result {
     let url = format!("http://{host}:{port}");
 
diff --git a/arrow-integration-testing/src/flight_client_scenarios/middleware.rs b/arrow-integration-testing/src/flight_client_scenarios/middleware.rs
index 3b71edf44..b826ad456 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/middleware.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/middleware.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Scenario for testing middleware.
+
 use arrow_flight::{
     flight_descriptor::DescriptorType, flight_service_client::FlightServiceClient, FlightDescriptor,
 };
@@ -24,6 +26,7 @@ use tonic::{Request, Status};
 type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 type Result<T = (), E = Error> = std::result::Result<T, E>;
 
+/// Run a scenario that tests middleware.
 pub async fn run_scenario(host: &str, port: u16) -> Result {
     let url = format!("http://{host}:{port}");
     let conn = tonic::transport::Endpoint::new(url)?.connect().await?;
diff --git a/arrow-integration-testing/src/flight_client_scenarios.rs b/arrow-integration-testing/src/flight_client_scenarios/mod.rs
similarity index 93%
rename from arrow-integration-testing/src/flight_client_scenarios.rs
rename to arrow-integration-testing/src/flight_client_scenarios/mod.rs
index 66cced5f4..c57944337 100644
--- a/arrow-integration-testing/src/flight_client_scenarios.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/mod.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Collection of utilities for testing the Flight client.
+
 pub mod auth_basic_proto;
 pub mod integration_test;
 pub mod middleware;
diff --git a/arrow-integration-testing/src/flight_server_scenarios/auth_basic_proto.rs b/arrow-integration-testing/src/flight_server_scenarios/auth_basic_proto.rs
index 20d868953..5462e5bd6 100644
--- a/arrow-integration-testing/src/flight_server_scenarios/auth_basic_proto.rs
+++ b/arrow-integration-testing/src/flight_server_scenarios/auth_basic_proto.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Basic auth test for the Flight server.
+
 use std::pin::Pin;
 use std::sync::Arc;
 
@@ -35,6 +37,7 @@ use prost::Message;
 
 use crate::{AUTH_PASSWORD, AUTH_USERNAME};
 
+/// Run a scenario that tests basic auth.
 pub async fn scenario_setup(port: u16) -> Result {
     let service = AuthBasicProtoScenarioImpl {
         username: AUTH_USERNAME.into(),
@@ -52,6 +55,7 @@ pub async fn scenario_setup(port: u16) -> Result {
     Ok(())
 }
 
+/// Scenario for testing basic auth.
 #[derive(Clone)]
 pub struct AuthBasicProtoScenarioImpl {
     username: Arc<str>,
diff --git a/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs b/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
index 76eb9d880..0c58fae93 100644
--- a/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -15,6 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Integration tests for the Flight server.
+
+use core::str;
 use std::collections::HashMap;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -42,6 +45,7 @@ type TonicStream<T> = Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>;
 type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 type Result<T = (), E = Error> = std::result::Result<T, E>;
 
+/// Run a scenario that tests integration testing.
 pub async fn scenario_setup(port: u16) -> Result {
     let addr = super::listen_on(port).await?;
 
@@ -65,6 +69,7 @@ struct IntegrationDataset {
     chunks: Vec<RecordBatch>,
 }
 
+/// Flight service implementation for integration testing
 #[derive(Clone, Default)]
 pub struct FlightServiceImpl {
     server_location: String,
@@ -100,13 +105,13 @@ impl FlightService for FlightServiceImpl {
     ) -> Result<Response<Self::DoGetStream>, Status> {
         let ticket = request.into_inner();
 
-        let key = String::from_utf8(ticket.ticket.to_vec())
+        let key = str::from_utf8(&ticket.ticket)
             .map_err(|e| Status::invalid_argument(format!("Invalid ticket: {e:?}")))?;
 
         let uploaded_chunks = self.uploaded_chunks.lock().await;
 
         let flight = uploaded_chunks
-            .get(&key)
+            .get(key)
             .ok_or_else(|| Status::not_found(format!("Could not find flight. {key}")))?;
 
         let options = arrow::ipc::writer::IpcWriteOptions::default();
diff --git a/arrow-integration-testing/src/flight_server_scenarios/middleware.rs b/arrow-integration-testing/src/flight_server_scenarios/middleware.rs
index e8d9c521b..6685d45df 100644
--- a/arrow-integration-testing/src/flight_server_scenarios/middleware.rs
+++ b/arrow-integration-testing/src/flight_server_scenarios/middleware.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Middleware test for the Flight server.
+
 use std::pin::Pin;
 
 use arrow_flight::{
@@ -31,6 +33,7 @@ type TonicStream<T> = Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>;
 type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 type Result<T = (), E = Error> = std::result::Result<T, E>;
 
+/// Run a scenario that tests middleware.
 pub async fn scenario_setup(port: u16) -> Result {
     let service = MiddlewareScenarioImpl {};
     let svc = FlightServiceServer::new(service);
@@ -44,6 +47,7 @@ pub async fn scenario_setup(port: u16) -> Result {
     Ok(())
 }
 
+/// Middleware interceptor for testing
 #[derive(Clone, Default)]
 pub struct MiddlewareScenarioImpl {}
 
diff --git a/arrow-integration-testing/src/flight_server_scenarios.rs b/arrow-integration-testing/src/flight_server_scenarios/mod.rs
similarity index 91%
rename from arrow-integration-testing/src/flight_server_scenarios.rs
rename to arrow-integration-testing/src/flight_server_scenarios/mod.rs
index 48d4e6045..3833e1c63 100644
--- a/arrow-integration-testing/src/flight_server_scenarios.rs
+++ b/arrow-integration-testing/src/flight_server_scenarios/mod.rs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Collection of utilities for testing the Flight server.
 use std::net::SocketAddr;
 
 use arrow_flight::{FlightEndpoint, Location, Ticket};
@@ -27,6 +28,7 @@ pub mod middleware;
 type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 type Result<T = (), E = Error> = std::result::Result<T, E>;
 
+/// Listen on a port and return the address
 pub async fn listen_on(port: u16) -> Result<SocketAddr> {
     let addr: SocketAddr = format!("0.0.0.0:{port}").parse()?;
 
@@ -36,6 +38,7 @@ pub async fn listen_on(port: u16) -> Result<SocketAddr> {
     Ok(addr)
 }
 
+/// Create a FlightEndpoint with a ticket and location
 pub fn endpoint(ticket: &str, location_uri: impl Into<String>) -> FlightEndpoint {
     FlightEndpoint {
         ticket: Some(Ticket {
diff --git a/arrow-integration-testing/src/lib.rs b/arrow-integration-testing/src/lib.rs
index 4ce7b06a1..ba8e3876c 100644
--- a/arrow-integration-testing/src/lib.rs
+++ b/arrow-integration-testing/src/lib.rs
@@ -17,6 +17,7 @@
 
 //! Common code used in the integration test binaries
 
+#![warn(missing_docs)]
 use serde_json::Value;
 
 use arrow::array::{Array, StructArray};
@@ -42,7 +43,9 @@ pub const AUTH_PASSWORD: &str = "flight";
 pub mod flight_client_scenarios;
 pub mod flight_server_scenarios;
 
+/// An Arrow file in JSON format
 pub struct ArrowFile {
+    /// The schema of the file
     pub schema: Schema,
     // we can evolve this into a concrete Arrow type
     // this is temporarily not being read from
@@ -51,12 +54,14 @@ pub struct ArrowFile {
 }
 
 impl ArrowFile {
+    /// Read a single [RecordBatch] from the file
     pub fn read_batch(&self, batch_num: usize) -> Result<RecordBatch> {
         let b = self.arrow_json["batches"].get(batch_num).unwrap();
         let json_batch: ArrowJsonBatch = serde_json::from_value(b.clone()).unwrap();
         record_batch_from_json(&self.schema, json_batch, Some(&self.dictionaries))
     }
 
+    /// Read all [RecordBatch]es from the file
     pub fn read_batches(&self) -> Result<Vec<RecordBatch>> {
         self.arrow_json["batches"]
             .as_array()
@@ -70,7 +75,7 @@ impl ArrowFile {
     }
 }
 
-// Canonicalize the names of map fields in a schema
+/// Canonicalize the names of map fields in a schema
 pub fn canonicalize_schema(schema: &Schema) -> Schema {
     let fields = schema
         .fields()
@@ -107,6 +112,7 @@ pub fn canonicalize_schema(schema: &Schema) -> Schema {
     Schema::new(fields).with_metadata(schema.metadata().clone())
 }
 
+/// Read an Arrow file in JSON format
 pub fn open_json_file(json_name: &str) -> Result<ArrowFile> {
     let json_file = File::open(json_name)?;
     let reader = BufReader::new(json_file);
@@ -157,10 +163,7 @@ pub fn read_gzip_json(version: &str, path: &str) -> ArrowJson {
     arrow_json
 }
 
-//
-// C Data Integration entrypoints
-//
-
+/// C Data Integration entrypoint to export the schema from a JSON file
 fn cdata_integration_export_schema_from_json(
     c_json_name: *const i8,
     out: *mut FFI_ArrowSchema,
@@ -173,6 +176,7 @@ fn cdata_integration_export_schema_from_json(
     Ok(())
 }
 
+/// C Data Integration entrypoint to export a batch from a JSON file
 fn cdata_integration_export_batch_from_json(
     c_json_name: *const i8,
     batch_num: c_int,
@@ -263,6 +267,7 @@ pub unsafe extern "C" fn arrow_rs_free_error(c_error: *mut i8) {
     }
 }
 
+/// A C-ABI for exporting an Arrow schema from a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_export_schema_from_json(
     c_json_name: *const i8,
@@ -272,6 +277,7 @@ pub extern "C" fn arrow_rs_cdata_integration_export_schema_from_json(
     result_to_c_error(&r)
 }
 
+/// A C-ABI to compare an Arrow schema against a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_import_schema_and_compare_to_json(
     c_json_name: *const i8,
@@ -281,6 +287,7 @@ pub extern "C" fn arrow_rs_cdata_integration_import_schema_and_compare_to_json(
     result_to_c_error(&r)
 }
 
+/// A C-ABI for exporting a RecordBatch from a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_export_batch_from_json(
     c_json_name: *const i8,
@@ -291,6 +298,7 @@ pub extern "C" fn arrow_rs_cdata_integration_export_batch_from_json(
     result_to_c_error(&r)
 }
 
+/// A C-ABI to compare a RecordBatch against a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_import_batch_and_compare_to_json(
     c_json_name: *const i8,
diff --git a/arrow/tests/array_cast.rs b/arrow/tests/array_cast.rs
index 0fd89cc2b..8f86cbeab 100644
--- a/arrow/tests/array_cast.rs
+++ b/arrow/tests/array_cast.rs
@@ -179,7 +179,7 @@ fn test_can_cast_types() {
 
 /// Create instances of arrays with varying types for cast tests
 fn get_arrays_of_all_types() -> Vec<ArrayRef> {
-    let tz_name = String::from("+08:00");
+    let tz_name = "+08:00";
     let binary_data: Vec<&[u8]> = vec![b"foo", b"bar"];
     vec![
         Arc::new(BinaryArray::from(binary_data.clone())),
@@ -238,9 +238,9 @@ fn get_arrays_of_all_types() -> Vec<ArrayRef> {
         Arc::new(TimestampMillisecondArray::from(vec![1000, 2000])),
         Arc::new(TimestampMicrosecondArray::from(vec![1000, 2000])),
         Arc::new(TimestampNanosecondArray::from(vec![1000, 2000])),
-        Arc::new(TimestampSecondArray::from(vec![1000, 2000]).with_timezone(tz_name.clone())),
-        Arc::new(TimestampMillisecondArray::from(vec![1000, 2000]).with_timezone(tz_name.clone())),
-        Arc::new(TimestampMicrosecondArray::from(vec![1000, 2000]).with_timezone(tz_name.clone())),
+        Arc::new(TimestampSecondArray::from(vec![1000, 2000]).with_timezone(tz_name)),
+        Arc::new(TimestampMillisecondArray::from(vec![1000, 2000]).with_timezone(tz_name)),
+        Arc::new(TimestampMicrosecondArray::from(vec![1000, 2000]).with_timezone(tz_name)),
         Arc::new(TimestampNanosecondArray::from(vec![1000, 2000]).with_timezone(tz_name)),
         Arc::new(Date32Array::from(vec![1000, 2000])),
         Arc::new(Date64Array::from(vec![1000, 2000])),
diff --git a/parquet/tests/arrow_reader/bad_data.rs b/parquet/tests/arrow_reader/bad_data.rs
index a73864070..e2975c17c 100644
--- a/parquet/tests/arrow_reader/bad_data.rs
+++ b/parquet/tests/arrow_reader/bad_data.rs
@@ -140,20 +140,23 @@ fn read_file(name: &str) -> Result<usize, ParquetError> {
 #[tokio::test]
 async fn bad_metadata_err() {
     use bytes::Bytes;
-    use parquet::arrow::async_reader::MetadataLoader;
+    use parquet::file::metadata::ParquetMetaDataReader;
 
     let metadata_buffer = Bytes::from_static(include_bytes!("bad_raw_metadata.bin"));
 
     let metadata_length = metadata_buffer.len();
 
     let mut reader = std::io::Cursor::new(&metadata_buffer);
-    let mut loader = MetadataLoader::load(&mut reader, metadata_length, None)
-        .await
-        .unwrap();
-    loader.load_page_index(false, false).await.unwrap();
-    loader.load_page_index(false, true).await.unwrap();
+    let mut loader = ParquetMetaDataReader::new();
+    loader.try_load(&mut reader, metadata_length).await.unwrap();
+    loader = loader.with_page_indexes(false);
+    loader.load_page_index(&mut reader).await.unwrap();
 
-    let err = loader.load_page_index(true, false).await.unwrap_err();
+    loader = loader.with_offset_indexes(true);
+    loader.load_page_index(&mut reader).await.unwrap();
+
+    loader = loader.with_column_indexes(true);
+    let err = loader.load_page_index(&mut reader).await.unwrap_err();
 
     assert_eq!(
         err.to_string(),
diff --git a/parquet/tests/arrow_reader/statistics.rs b/parquet/tests/arrow_reader/statistics.rs
index 384be83d3..29adbbb4f 100644
--- a/parquet/tests/arrow_reader/statistics.rs
+++ b/parquet/tests/arrow_reader/statistics.rs
@@ -22,6 +22,7 @@ use std::default::Default;
 use std::fs::File;
 use std::sync::Arc;
 
+use super::make_test_file_rg;
 use super::{struct_array, Scenario};
 use arrow::compute::kernels::cast_utils::Parser;
 use arrow::datatypes::{
@@ -37,16 +38,17 @@ use arrow_array::{
     TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt16Array,
     UInt32Array, UInt64Array, UInt8Array,
 };
-use arrow_schema::{DataType, Field, Schema, TimeUnit};
+use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
 use half::f16;
 use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
 use parquet::arrow::arrow_reader::{
     ArrowReaderBuilder, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
 };
 use parquet::arrow::ArrowWriter;
+use parquet::file::metadata::{ColumnChunkMetaData, RowGroupMetaData};
 use parquet::file::properties::{EnabledStatistics, WriterProperties};
-
-use super::make_test_file_rg;
+use parquet::file::statistics::{Statistics, ValueStatistics};
+use parquet::schema::types::{SchemaDescPtr, SchemaDescriptor};
 
 #[derive(Debug, Default, Clone)]
 struct Int64Case {
@@ -2139,6 +2141,65 @@ async fn test_missing_statistics() {
     .run();
 }
 
+#[test]
+fn missing_null_counts_as_zero() {
+    let min = None;
+    let max = None;
+    let distinct_count = None;
+    let null_count = None; // NB: no null count
+    let is_min_max_deprecated = false;
+    let stats = Statistics::Boolean(ValueStatistics::new(
+        min,
+        max,
+        distinct_count,
+        null_count,
+        is_min_max_deprecated,
+    ));
+    let (arrow_schema, parquet_schema) = bool_arrow_and_parquet_schema();
+
+    let column_chunk = ColumnChunkMetaData::builder(parquet_schema.column(0))
+        .set_statistics(stats)
+        .build()
+        .unwrap();
+    let metadata = RowGroupMetaData::builder(parquet_schema.clone())
+        .set_column_metadata(vec![column_chunk])
+        .build()
+        .unwrap();
+
+    let converter = StatisticsConverter::try_new("b", &arrow_schema, &parquet_schema).unwrap();
+
+    // by default null count should be 0
+    assert_eq!(
+        converter.row_group_null_counts([&metadata]).unwrap(),
+        UInt64Array::from_iter(vec![Some(0)])
+    );
+
+    // if we disable missing null counts as zero flag null count will be None
+    let converter = converter.with_missing_null_counts_as_zero(false);
+    assert_eq!(
+        converter.row_group_null_counts([&metadata]).unwrap(),
+        UInt64Array::from_iter(vec![None])
+    );
+}
+
+/// return an Arrow schema and corresponding Parquet SchemaDescriptor for
+/// a schema with a single boolean column "b"
+fn bool_arrow_and_parquet_schema() -> (SchemaRef, SchemaDescPtr) {
+    let arrow_schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Boolean, true)]));
+    use parquet::schema::types::Type as ParquetType;
+    let parquet_schema = ParquetType::group_type_builder("schema")
+        .with_fields(vec![Arc::new(
+            ParquetType::primitive_type_builder("a", parquet::basic::Type::INT32)
+                .build()
+                .unwrap(),
+        )])
+        .build()
+        .unwrap();
+
+    let parquet_schema = Arc::new(SchemaDescriptor::new(Arc::new(parquet_schema)));
+    (arrow_schema, parquet_schema)
+}
+
 /////// NEGATIVE TESTS ///////
 // column not found
 #[tokio::test]