Repository: apache/arrow-rs
Pull request #6986 (2025-01-15 12:48:33)

Add example reading data from an `mmap`ed IPC file
# Which issue does this PR close?

- Closes https://github.com/apache/arrow-rs/issues/6709

# Rationale for this change
 
Reading arrow IPC files without copying is a key format feature, but it is hard to understand how to make this happen today

# What changes are included in this PR?

1. Add an example of how to use `mmap` with `FileDecoder`

# Are there any user-facing changes?
Example

Potential follow ons (TODO):
1. Create a `Buffer::from_owner` as suggested by @tustvold  in https://github.com/apache/arrow-rs/issues/6709#issuecomment-2591285989
2. Maybe add `IPCBufferDecoder` or something similar into the actual crate API


Patch:
diff --git a/.github/workflows/parquet.yml b/.github/workflows/parquet.yml
index 19503fde7..4c46fde19 100644
--- a/.github/workflows/parquet.yml
+++ b/.github/workflows/parquet.yml
@@ -97,6 +97,8 @@ jobs:
         run: cargo check -p parquet --no-default-features
       - name: Check compilation --no-default-features --features arrow
         run: cargo check -p parquet --no-default-features --features arrow
+      - name: Check compilation --no-default-features --features simdutf8
+        run: cargo check -p parquet --no-default-features --features simdutf8
       - name: Check compilation --no-default-features --all-features
         run: cargo check -p parquet --all-features
       - name: Check compilation --all-targets
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index 5cf208c8e..4dcd56156 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -683,6 +683,10 @@ pub fn read_footer_length(buf: [u8; 10]) -> Result<usize, ArrowError> {
 ///
 /// For a higher-level interface see [`FileReader`]
 ///
+/// For an example of using this API with `mmap` see the [`zero_copy_ipc`] example.
+///
+/// [`zero_copy_ipc`]: https://github.com/apache/arrow-rs/blob/main/arrow/examples/zero_copy_ipc.rs
+///
 /// ```
 /// # use std::sync::Arc;
 /// # use arrow_array::*;
@@ -994,6 +998,10 @@ impl FileReaderBuilder {
 }
 
 /// Arrow File reader
+///
+/// For an example creating Arrays with memory mapped (`mmap`) files see the [`zero_copy_ipc`] example.
+///
+/// [`zero_copy_ipc`]: https://github.com/apache/arrow-rs/blob/main/arrow/examples/zero_copy_ipc.rs
 pub struct FileReader<R> {
     /// File reader that supports reading and seeking
     reader: R,
diff --git a/arrow-schema/src/datatype.rs b/arrow-schema/src/datatype.rs
index 7cd53b13c..23bd2e2cf 100644
--- a/arrow-schema/src/datatype.rs
+++ b/arrow-schema/src/datatype.rs
@@ -552,16 +552,21 @@ impl DataType {
         matches!(self, Int16 | Int32 | Int64)
     }
 
-    /// Returns true if this type is nested (List, FixedSizeList, LargeList, Struct, Union,
+    /// Returns true if this type is nested (List, FixedSizeList, LargeList, ListView. LargeListView, Struct, Union,
     /// or Map), or a dictionary of a nested type
     #[inline]
     pub fn is_nested(&self) -> bool {
         use DataType::*;
         match self {
             Dictionary(_, v) => DataType::is_nested(v.as_ref()),
-            List(_) | FixedSizeList(_, _) | LargeList(_) | Struct(_) | Union(_, _) | Map(_, _) => {
-                true
-            }
+            List(_)
+            | FixedSizeList(_, _)
+            | LargeList(_)
+            | ListView(_)
+            | LargeListView(_)
+            | Struct(_)
+            | Union(_, _)
+            | Map(_, _) => true,
             _ => false,
         }
     }
@@ -578,7 +583,9 @@ impl DataType {
     pub fn equals_datatype(&self, other: &DataType) -> bool {
         match (&self, other) {
             (DataType::List(a), DataType::List(b))
-            | (DataType::LargeList(a), DataType::LargeList(b)) => {
+            | (DataType::LargeList(a), DataType::LargeList(b))
+            | (DataType::ListView(a), DataType::ListView(b))
+            | (DataType::LargeListView(a), DataType::LargeListView(b)) => {
                 a.is_nullable() == b.is_nullable() && a.data_type().equals_datatype(b.data_type())
             }
             (DataType::FixedSizeList(a, a_size), DataType::FixedSizeList(b, b_size)) => {
@@ -725,7 +732,9 @@ impl DataType {
     pub fn contains(&self, other: &DataType) -> bool {
         match (self, other) {
             (DataType::List(f1), DataType::List(f2))
-            | (DataType::LargeList(f1), DataType::LargeList(f2)) => f1.contains(f2),
+            | (DataType::LargeList(f1), DataType::LargeList(f2))
+            | (DataType::ListView(f1), DataType::ListView(f2))
+            | (DataType::LargeListView(f1), DataType::LargeListView(f2)) => f1.contains(f2),
             (DataType::FixedSizeList(f1, s1), DataType::FixedSizeList(f2, s2)) => {
                 s1 == s2 && f1.contains(f2)
             }
@@ -1006,11 +1015,16 @@ mod tests {
     #[test]
     fn test_nested() {
         let list = DataType::List(Arc::new(Field::new("foo", DataType::Utf8, true)));
+        let list_view = DataType::ListView(Arc::new(Field::new("foo", DataType::Utf8, true)));
+        let large_list_view =
+            DataType::LargeListView(Arc::new(Field::new("foo", DataType::Utf8, true)));
 
         assert!(!DataType::is_nested(&DataType::Boolean));
         assert!(!DataType::is_nested(&DataType::Int32));
         assert!(!DataType::is_nested(&DataType::Utf8));
         assert!(DataType::is_nested(&list));
+        assert!(DataType::is_nested(&list_view));
+        assert!(DataType::is_nested(&large_list_view));
 
         assert!(!DataType::is_nested(&DataType::Dictionary(
             Box::new(DataType::Int32),
diff --git a/arrow-select/src/interleave.rs b/arrow-select/src/interleave.rs
index 4a47017b7..3557bda8f 100644
--- a/arrow-select/src/interleave.rs
+++ b/arrow-select/src/interleave.rs
@@ -24,7 +24,9 @@ use arrow_array::types::*;
 use arrow_array::*;
 use arrow_buffer::{ArrowNativeType, MutableBuffer, NullBuffer, NullBufferBuilder, OffsetBuffer};
 use arrow_data::transform::MutableArrayData;
+use arrow_data::ByteView;
 use arrow_schema::{ArrowError, DataType};
+use std::collections::HashMap;
 use std::sync::Arc;
 
 macro_rules! primitive_helper {
@@ -97,6 +99,8 @@ pub fn interleave(
         DataType::LargeUtf8 => interleave_bytes::<LargeUtf8Type>(values, indices),
         DataType::Binary => interleave_bytes::<BinaryType>(values, indices),
         DataType::LargeBinary => interleave_bytes::<LargeBinaryType>(values, indices),
+        DataType::BinaryView => interleave_views::<BinaryViewType>(values, indices),
+        DataType::Utf8View => interleave_views::<StringViewType>(values, indices),
         DataType::Dictionary(k, _) => downcast_integer! {
             k.as_ref() => (dict_helper, values, indices),
             _ => unreachable!("illegal dictionary key type {k}")
@@ -231,6 +235,41 @@ fn interleave_dictionaries<K: ArrowDictionaryKeyType>(
     Ok(Arc::new(array))
 }
 
+fn interleave_views<T: ByteViewType>(
+    values: &[&dyn Array],
+    indices: &[(usize, usize)],
+) -> Result<ArrayRef, ArrowError> {
+    let interleaved = Interleave::<'_, GenericByteViewArray<T>>::new(values, indices);
+    let mut views_builder = BufferBuilder::new(indices.len());
+    let mut buffers = Vec::new();
+
+    // (input array_index, input buffer_index) -> output buffer_index
+    let mut buffer_lookup: HashMap<(usize, u32), u32> = HashMap::new();
+    for (array_idx, value_idx) in indices {
+        let array = interleaved.arrays[*array_idx];
+        let raw_view = array.views().get(*value_idx).unwrap();
+        let view_len = *raw_view as u32;
+        if view_len <= 12 {
+            views_builder.append(*raw_view);
+            continue;
+        }
+        // value is big enough to be in a variadic buffer
+        let view = ByteView::from(*raw_view);
+        let new_buffer_idx: &mut u32 = buffer_lookup
+            .entry((*array_idx, view.buffer_index))
+            .or_insert_with(|| {
+                buffers.push(array.data_buffers()[view.buffer_index as usize].clone());
+                (buffers.len() - 1) as u32
+            });
+        views_builder.append(view.with_buffer_index(*new_buffer_idx).into());
+    }
+
+    let array = unsafe {
+        GenericByteViewArray::<T>::new_unchecked(views_builder.into(), buffers, interleaved.nulls)
+    };
+    Ok(Arc::new(array))
+}
+
 /// Fallback implementation of interleave using [`MutableArrayData`]
 fn interleave_fallback(
     values: &[&dyn Array],
@@ -461,4 +500,209 @@ mod tests {
             DictionaryArray::<Int32Type>::from_iter(vec![Some("0"), Some("1"), Some("2"), None]);
         assert_eq!(array.as_ref(), &expected)
     }
+
+    #[test]
+    fn test_interleave_views() {
+        let values = StringArray::from_iter_values([
+            "hello",
+            "world_long_string_not_inlined",
+            "foo",
+            "bar",
+            "baz",
+        ]);
+        let view_a = StringViewArray::from(&values);
+
+        let values = StringArray::from_iter_values([
+            "test",
+            "data",
+            "more_long_string_not_inlined",
+            "views",
+            "here",
+        ]);
+        let view_b = StringViewArray::from(&values);
+
+        let indices = &[
+            (0, 2), // "foo"
+            (1, 0), // "test"
+            (0, 4), // "baz"
+            (1, 3), // "views"
+            (0, 1), // "world_long_string_not_inlined"
+        ];
+
+        // Test specialized implementation
+        let values = interleave(&[&view_a, &view_b], indices).unwrap();
+        let result = values.as_string_view();
+        assert_eq!(result.data_buffers().len(), 1);
+
+        let fallback = interleave_fallback(&[&view_a, &view_b], indices).unwrap();
+        let fallback_result = fallback.as_string_view();
+        // note that fallback_result has 2 buffers, but only one long enough string to warrant a buffer
+        assert_eq!(fallback_result.data_buffers().len(), 2);
+
+        // Convert to strings for easier assertion
+        let collected: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
+
+        let fallback_collected: Vec<_> = fallback_result
+            .iter()
+            .map(|x| x.map(|s| s.to_string()))
+            .collect();
+
+        assert_eq!(&collected, &fallback_collected);
+
+        assert_eq!(
+            &collected,
+            &[
+                Some("foo".to_string()),
+                Some("test".to_string()),
+                Some("baz".to_string()),
+                Some("views".to_string()),
+                Some("world_long_string_not_inlined".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn test_interleave_views_with_nulls() {
+        let values = StringArray::from_iter([
+            Some("hello"),
+            None,
+            Some("foo_long_string_not_inlined"),
+            Some("bar"),
+            None,
+        ]);
+        let view_a = StringViewArray::from(&values);
+
+        let values = StringArray::from_iter([
+            Some("test"),
+            Some("data_long_string_not_inlined"),
+            None,
+            None,
+            Some("here"),
+        ]);
+        let view_b = StringViewArray::from(&values);
+
+        let indices = &[
+            (0, 1), // null
+            (1, 2), // null
+            (0, 2), // "foo_long_string_not_inlined"
+            (1, 3), // null
+            (0, 4), // null
+        ];
+
+        // Test specialized implementation
+        let values = interleave(&[&view_a, &view_b], indices).unwrap();
+        let result = values.as_string_view();
+        assert_eq!(result.data_buffers().len(), 1);
+
+        let fallback = interleave_fallback(&[&view_a, &view_b], indices).unwrap();
+        let fallback_result = fallback.as_string_view();
+
+        // Convert to strings for easier assertion
+        let collected: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
+
+        let fallback_collected: Vec<_> = fallback_result
+            .iter()
+            .map(|x| x.map(|s| s.to_string()))
+            .collect();
+
+        assert_eq!(&collected, &fallback_collected);
+
+        assert_eq!(
+            &collected,
+            &[
+                None,
+                None,
+                Some("foo_long_string_not_inlined".to_string()),
+                None,
+                None,
+            ]
+        );
+    }
+
+    #[test]
+    fn test_interleave_views_multiple_buffers() {
+        let str1 = "very_long_string_from_first_buffer".as_bytes();
+        let str2 = "very_long_string_from_second_buffer".as_bytes();
+        let buffer1 = str1.to_vec().into();
+        let buffer2 = str2.to_vec().into();
+
+        let view1 = ByteView::new(str1.len() as u32, &str1[..4])
+            .with_buffer_index(0)
+            .with_offset(0)
+            .as_u128();
+        let view2 = ByteView::new(str2.len() as u32, &str2[..4])
+            .with_buffer_index(1)
+            .with_offset(0)
+            .as_u128();
+        let view_a =
+            StringViewArray::try_new(vec![view1, view2].into(), vec![buffer1, buffer2], None)
+                .unwrap();
+
+        let str3 = "another_very_long_string_buffer_three".as_bytes();
+        let str4 = "different_long_string_in_buffer_four".as_bytes();
+        let buffer3 = str3.to_vec().into();
+        let buffer4 = str4.to_vec().into();
+
+        let view3 = ByteView::new(str3.len() as u32, &str3[..4])
+            .with_buffer_index(0)
+            .with_offset(0)
+            .as_u128();
+        let view4 = ByteView::new(str4.len() as u32, &str4[..4])
+            .with_buffer_index(1)
+            .with_offset(0)
+            .as_u128();
+        let view_b =
+            StringViewArray::try_new(vec![view3, view4].into(), vec![buffer3, buffer4], None)
+                .unwrap();
+
+        let indices = &[
+            (0, 0), // String from first buffer of array A
+            (1, 0), // String from first buffer of array B
+            (0, 1), // String from second buffer of array A
+            (1, 1), // String from second buffer of array B
+            (0, 0), // String from first buffer of array A again
+            (1, 1), // String from second buffer of array B again
+        ];
+
+        // Test interleave
+        let values = interleave(&[&view_a, &view_b], indices).unwrap();
+        let result = values.as_string_view();
+
+        assert_eq!(
+            result.data_buffers().len(),
+            4,
+            "Expected four buffers (two from each input array)"
+        );
+
+        let result_strings: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
+        assert_eq!(
+            result_strings,
+            vec![
+                Some("very_long_string_from_first_buffer".to_string()),
+                Some("another_very_long_string_buffer_three".to_string()),
+                Some("very_long_string_from_second_buffer".to_string()),
+                Some("different_long_string_in_buffer_four".to_string()),
+                Some("very_long_string_from_first_buffer".to_string()),
+                Some("different_long_string_in_buffer_four".to_string()),
+            ]
+        );
+
+        let views = result.views();
+        let buffer_indices: Vec<_> = views
+            .iter()
+            .map(|raw_view| ByteView::from(*raw_view).buffer_index)
+            .collect();
+
+        assert_eq!(
+            buffer_indices,
+            vec![
+                0, // First buffer from array A
+                1, // First buffer from array B
+                2, // Second buffer from array A
+                3, // Second buffer from array B
+                0, // First buffer from array A (reused)
+                3, // Second buffer from array B (reused)
+            ]
+        );
+    }
 }
diff --git a/arrow/Cargo.toml b/arrow/Cargo.toml
index 76119ec4a..1b01dcd25 100644
--- a/arrow/Cargo.toml
+++ b/arrow/Cargo.toml
@@ -87,6 +87,9 @@ criterion = { version = "0.5", default-features = false }
 half = { version = "2.1", default-features = false }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
 serde = { version = "1.0", default-features = false, features = ["derive"] }
+# used in examples
+memmap2 = "0.9.3"
+bytes = "1.9"
 
 [build-dependencies]
 
@@ -105,6 +108,11 @@ name = "read_csv_infer_schema"
 required-features = ["prettyprint", "csv"]
 path = "./examples/read_csv_infer_schema.rs"
 
+[[example]]
+name = "zero_copy_ipc"
+required-features = ["prettyprint"]
+path = "examples/zero_copy_ipc.rs"
+
 [[bench]]
 name = "aggregate_kernels"
 harness = false
diff --git a/arrow/examples/README.md b/arrow/examples/README.md
index 87aa6ee04..9d896cd11 100644
--- a/arrow/examples/README.md
+++ b/arrow/examples/README.md
@@ -24,5 +24,6 @@
 - [`dynamic_types.rs`](dynamic_types.rs): Dealing with mixed types dynamically at runtime
 - [`read_csv.rs`](read_csv.rs): Reading CSV files with explicit schema, pretty printing Arrays
 - [`read_csv_infer_schema.rs`](read_csv_infer_schema.rs): Reading CSV files, pretty printing Arrays
+- [`zero_copy_ipc`](zero_copy_ipc): Zero copy read of Arrow IPC file using `mmap`
 - [`tensor_builder.rs`](tensor_builder.rs): Using tensor builder
 - [`version.rs`](version.rs): Print the arrow version and exit
diff --git a/arrow/examples/zero_copy_ipc.rs b/arrow/examples/zero_copy_ipc.rs
new file mode 100644
index 000000000..15fc477c5
--- /dev/null
+++ b/arrow/examples/zero_copy_ipc.rs
@@ -0,0 +1,157 @@
+// 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.
+
+//! Example of reading arrow IPC files and streams using "zero copy" API
+//!
+//! Zero copy in this case means the Arrow arrays refer directly to a user
+//! provided buffer or memory region.
+
+use arrow::array::{record_batch, RecordBatch};
+use arrow::error::Result;
+use arrow_buffer::Buffer;
+use arrow_cast::pretty::pretty_format_batches;
+use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::reader::{read_footer_length, FileDecoder};
+use arrow_ipc::writer::FileWriter;
+use arrow_ipc::{root_as_footer, Block};
+use std::path::PathBuf;
+use std::sync::Arc;
+
+/// This example shows how to read data from an Arrow IPC file without copying
+/// using `mmap` and the [`FileDecoder`] API
+fn main() {
+    // Create a temporary file with 3 record batches
+    let ipc_path = ipc_file();
+    // Open the file and memory map it using the mmap2 crate:
+    let ipc_file = std::fs::File::open(&ipc_path.path).expect("failed to open file");
+    let mmap = unsafe { memmap2::Mmap::map(&ipc_file).expect("failed to mmap file") };
+
+    // Convert the mmap region to an Arrow `Buffer` to back the arrow arrays. We
+    // do this by first creating a `bytes::Bytes` (which is zero copy) and then
+    // creating a Buffer from the `Bytes` (which is also zero copy)
+    let bytes = bytes::Bytes::from_owner(mmap);
+    let buffer = Buffer::from(bytes);
+
+    // Now, use the FileDecoder API (wrapped by `IPCBufferDecoder` for
+    // convenience) to crate Arrays re-using the data in the underlying buffer
+    let decoder = IPCBufferDecoder::new(buffer);
+    assert_eq!(decoder.num_batches(), 3);
+
+    // Create the Arrays and print them
+    for i in 0..decoder.num_batches() {
+        let batch = decoder.get_batch(i).unwrap().expect("failed to read batch");
+        assert_eq!(3, batch.num_rows());
+        println!("Batch {i}\n{}", pretty_format_batches(&[batch]).unwrap());
+    }
+}
+
+/// Return 3 [`RecordBatch`]es with a single column of type `Int32`
+fn example_data() -> Vec<RecordBatch> {
+    vec![
+        record_batch!(("my_column", Int32, [1, 2, 3])).unwrap(),
+        record_batch!(("my_column", Int32, [4, 5, 6])).unwrap(),
+        record_batch!(("my_column", Int32, [7, 8, 9])).unwrap(),
+    ]
+}
+
+/// Return a temporary file that contains an IPC file with 3 [`RecordBatch`]es
+fn ipc_file() -> TempFile {
+    let path = PathBuf::from("example.arrow");
+    let file = std::fs::File::create(&path).unwrap();
+    let data = example_data();
+    let mut writer = FileWriter::try_new(file, &data[0].schema()).unwrap();
+    for batch in &data {
+        writer.write(batch).unwrap();
+    }
+    writer.finish().unwrap();
+    TempFile { path }
+}
+
+/// Incrementally decodes [`RecordBatch`]es from an IPC file stored in a Arrow
+/// [`Buffer`] using the [`FileDecoder`] API.
+///
+/// This is a wrapper around the example in the `FileDecoder` which handles the
+/// low level interaction with the Arrow IPC format.
+struct IPCBufferDecoder {
+    /// Memory (or memory mapped) Buffer with the data
+    buffer: Buffer,
+    /// Decoder that reads Arrays that refers to the underlying buffers
+    decoder: FileDecoder,
+    /// Location of the batches within the buffer
+    batches: Vec<Block>,
+}
+
+impl IPCBufferDecoder {
+    fn new(buffer: Buffer) -> Self {
+        let trailer_start = buffer.len() - 10;
+        let footer_len = read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
+        let footer = root_as_footer(&buffer[trailer_start - footer_len..trailer_start]).unwrap();
+
+        let schema = fb_to_schema(footer.schema().unwrap());
+
+        let mut decoder = FileDecoder::new(Arc::new(schema), footer.version());
+
+        // Read dictionaries
+        for block in footer.dictionaries().iter().flatten() {
+            let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+            let data = buffer.slice_with_length(block.offset() as _, block_len);
+            decoder.read_dictionary(block, &data).unwrap();
+        }
+
+        // convert to Vec from the flatbuffers Vector to avoid having a direct dependency on flatbuffers
+        let batches = footer
+            .recordBatches()
+            .map(|b| b.iter().copied().collect())
+            .unwrap_or_default();
+
+        Self {
+            buffer,
+            decoder,
+            batches,
+        }
+    }
+
+    /// Return the number of [`RecordBatch`]es in this buffer
+    fn num_batches(&self) -> usize {
+        self.batches.len()
+    }
+
+    /// Return the [`RecordBatch`] at message index `i`.
+    ///
+    /// This may return `None` if the IPC message was None
+    fn get_batch(&self, i: usize) -> Result<Option<RecordBatch>> {
+        let block = &self.batches[i];
+        let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+        let data = self
+            .buffer
+            .slice_with_length(block.offset() as _, block_len);
+        self.decoder.read_record_batch(block, &data)
+    }
+}
+
+/// This structure deletes the file when it is dropped
+struct TempFile {
+    path: PathBuf,
+}
+
+impl Drop for TempFile {
+    fn drop(&mut self) {
+        if let Err(e) = std::fs::remove_file(&self.path) {
+            println!("Error deleting '{:?}': {:?}", self.path, e);
+        }
+    }
+}
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index fa5412c45..2c2e27ea4 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -1058,7 +1058,7 @@ impl TryFrom<Blob> for ObjectMeta {
         Ok(Self {
             location: Path::parse(value.name)?,
             last_modified: value.properties.last_modified,
-            size: value.properties.content_length as usize,
+            size: value.properties.content_length,
             e_tag: value.properties.e_tag,
             version: None, // For consistency with S3 and GCP which don't include this
         })
diff --git a/object_store/src/chunked.rs b/object_store/src/chunked.rs
index 4998e9f2a..2bb30b905 100644
--- a/object_store/src/chunked.rs
+++ b/object_store/src/chunked.rs
@@ -44,7 +44,7 @@ use crate::{PutPayload, Result};
 #[derive(Debug)]
 pub struct ChunkedStore {
     inner: Arc<dyn ObjectStore>,
-    chunk_size: usize,
+    chunk_size: usize, // chunks are in memory, so we use usize not u64
 }
 
 impl ChunkedStore {
@@ -138,7 +138,7 @@ impl ObjectStore for ChunkedStore {
         })
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         self.inner.get_range(location, range).await
     }
 
@@ -203,8 +203,8 @@ mod tests {
 
             let mut remaining = 1001;
             while let Some(next) = s.next().await {
-                let size = next.unwrap().len();
-                let expected = remaining.min(chunk_size);
+                let size = next.unwrap().len() as u64;
+                let expected = remaining.min(chunk_size as u64);
                 assert_eq!(size, expected);
                 remaining -= expected;
             }
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 57aca8956..f252dd9c2 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -67,9 +67,9 @@ impl<T: GetClient> GetClientExt for T {
 
 struct ContentRange {
     /// The range of the object returned
-    range: Range<usize>,
+    range: Range<u64>,
     /// The total size of the object being requested
-    size: usize,
+    size: u64,
 }
 
 impl ContentRange {
@@ -84,7 +84,7 @@ impl ContentRange {
         let (start_s, end_s) = range.split_once('-')?;
 
         let start = start_s.parse().ok()?;
-        let end: usize = end_s.parse().ok()?;
+        let end: u64 = end_s.parse().ok()?;
 
         Some(Self {
             size,
@@ -140,8 +140,8 @@ enum GetResultError {
 
     #[error("Requested {expected:?}, got {actual:?}")]
     UnexpectedRange {
-        expected: Range<usize>,
-        actual: Range<usize>,
+        expected: Range<u64>,
+        actual: Range<u64>,
     },
 }
 
diff --git a/object_store/src/client/s3.rs b/object_store/src/client/s3.rs
index 7fe956b23..a2221fbbc 100644
--- a/object_store/src/client/s3.rs
+++ b/object_store/src/client/s3.rs
@@ -66,7 +66,7 @@ pub struct ListPrefix {
 #[serde(rename_all = "PascalCase")]
 pub struct ListContents {
     pub key: String,
-    pub size: usize,
+    pub size: u64,
     pub last_modified: DateTime<Utc>,
     #[serde(rename = "ETag")]
     pub e_tag: Option<String>,
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index 41e6464c1..9983fdff5 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -420,7 +420,7 @@ impl MultiStatusResponse {
         })?)
     }
 
-    fn size(&self) -> Result<usize> {
+    fn size(&self) -> Result<u64> {
         let size = self
             .prop_stat
             .prop
@@ -462,7 +462,7 @@ pub(crate) struct Prop {
     last_modified: DateTime<Utc>,
 
     #[serde(rename = "getcontentlength")]
-    content_length: Option<usize>,
+    content_length: Option<u64>,
 
     #[serde(rename = "resourcetype")]
     resource_type: ResourceType,
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
index 20e95fddc..25a929459 100644
--- a/object_store/src/integration.rs
+++ b/object_store/src/integration.rs
@@ -112,7 +112,7 @@ pub async fn put_get_delete_list(storage: &DynObjectStore) {
     let range_result = storage.get_range(&location, range.clone()).await;
 
     let bytes = range_result.unwrap();
-    assert_eq!(bytes, data.slice(range.clone()));
+    assert_eq!(bytes, data.slice(range.start as usize..range.end as usize));
 
     let opts = GetOptions {
         range: Some(GetRange::Bounded(2..5)),
@@ -190,11 +190,11 @@ pub async fn put_get_delete_list(storage: &DynObjectStore) {
     let ranges = vec![0..1, 2..3, 0..5];
     let bytes = storage.get_ranges(&location, &ranges).await.unwrap();
     for (range, bytes) in ranges.iter().zip(bytes) {
-        assert_eq!(bytes, data.slice(range.clone()))
+        assert_eq!(bytes, data.slice(range.start as usize..range.end as usize));
     }
 
     let head = storage.head(&location).await.unwrap();
-    assert_eq!(head.size, data.len());
+    assert_eq!(head.size, data.len() as u64);
 
     storage.delete(&location).await.unwrap();
 
@@ -934,7 +934,7 @@ pub async fn list_with_delimiter(storage: &DynObjectStore) {
     let object = &result.objects[0];
 
     assert_eq!(object.location, expected_location);
-    assert_eq!(object.size, data.len());
+    assert_eq!(object.size, data.len() as u64);
 
     // ==================== check: prefix-list `mydb/wb/000/000/001` (partial filename doesn't match) ====================
     let prefix = Path::from("mydb/wb/000/000/001");
@@ -1085,7 +1085,7 @@ pub async fn multipart(storage: &dyn ObjectStore, multipart: &dyn MultipartStore
         .unwrap();
 
     let meta = storage.head(&path).await.unwrap();
-    assert_eq!(meta.size, chunk_size * 2);
+    assert_eq!(meta.size, chunk_size as u64 * 2);
 
     // Empty case
     let path = Path::from("test_empty_multipart");
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 53eda5a82..cffcbbdd4 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -234,7 +234,7 @@
 //!
 //! // Buffer the entire object in memory
 //! let object: Bytes = result.bytes().await.unwrap();
-//! assert_eq!(object.len(), meta.size);
+//! assert_eq!(object.len() as u64, meta.size);
 //!
 //! // Alternatively stream the bytes from object storage
 //! let stream = object_store.get(&path).await.unwrap().into_stream();
@@ -630,7 +630,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
     /// in the given byte range.
     ///
     /// See [`GetRange::Bounded`] for more details on how `range` gets interpreted
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let options = GetOptions {
             range: Some(range.into()),
             ..Default::default()
@@ -640,7 +640,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
 
     /// Return the bytes that are stored at the specified location
     /// in the given byte ranges
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         coalesce_ranges(
             ranges,
             |range| self.get_range(location, range),
@@ -820,14 +820,14 @@ macro_rules! as_ref_impl {
                 self.as_ref().get_opts(location, options).await
             }
 
-            async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+            async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
                 self.as_ref().get_range(location, range).await
             }
 
             async fn get_ranges(
                 &self,
                 location: &Path,
-                ranges: &[Range<usize>],
+                ranges: &[Range<u64>],
             ) -> Result<Vec<Bytes>> {
                 self.as_ref().get_ranges(location, ranges).await
             }
@@ -903,8 +903,10 @@ pub struct ObjectMeta {
     pub location: Path,
     /// The last modified time
     pub last_modified: DateTime<Utc>,
-    /// The size in bytes of the object
-    pub size: usize,
+    /// The size in bytes of the object.
+    ///
+    /// Note this is not `usize` as `object_store` supports 32-bit architectures such as WASM
+    pub size: u64,
     /// The unique identifier for the object
     ///
     /// <https://datatracker.ietf.org/doc/html/rfc9110#name-etag>
@@ -1019,7 +1021,9 @@ pub struct GetResult {
     /// The [`ObjectMeta`] for this object
     pub meta: ObjectMeta,
     /// The range of bytes returned by this request
-    pub range: Range<usize>,
+    ///
+    /// Note this is not `usize` as `object_store` supports 32-bit architectures such as WASM
+    pub range: Range<u64>,
     /// Additional object attributes
     pub attributes: Attributes,
 }
@@ -1060,7 +1064,11 @@ impl GetResult {
                             path: path.clone(),
                         })?;
 
-                    let mut buffer = Vec::with_capacity(len);
+                    let mut buffer = if let Ok(len) = len.try_into() {
+                        Vec::with_capacity(len)
+                    } else {
+                        Vec::new()
+                    };
                     file.take(len as _)
                         .read_to_end(&mut buffer)
                         .map_err(|source| local::Error::UnableToReadBytes { source, path })?;
diff --git a/object_store/src/limit.rs b/object_store/src/limit.rs
index 77f72a0e1..330a0da07 100644
--- a/object_store/src/limit.rs
+++ b/object_store/src/limit.rs
@@ -117,12 +117,12 @@ impl<T: ObjectStore> ObjectStore for LimitStore<T> {
         Ok(permit_get_result(r, permit))
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let _permit = self.semaphore.acquire().await.unwrap();
         self.inner.get_range(location, range).await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let _permit = self.semaphore.acquire().await.unwrap();
         self.inner.get_ranges(location, ranges).await
     }
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 364026459..65e87f982 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -21,7 +21,7 @@ use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
 use std::ops::Range;
 use std::sync::Arc;
 use std::time::SystemTime;
-use std::{collections::BTreeSet, convert::TryFrom, io};
+use std::{collections::BTreeSet, io};
 use std::{collections::VecDeque, path::PathBuf};
 
 use async_trait::async_trait;
@@ -44,12 +44,6 @@ use crate::{
 /// A specialized `Error` for filesystem object store-related errors
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
-    #[error("File size for {} did not fit in a usize: {}", path, source)]
-    FileSizeOverflowedUsize {
-        source: std::num::TryFromIntError,
-        path: String,
-    },
-
     #[error("Unable to walk dir: {}", source)]
     UnableToWalkDir { source: walkdir::Error },
 
@@ -83,8 +77,8 @@ pub(crate) enum Error {
     #[error("Out of range of file {}, expected: {}, actual: {}", path.display(), expected, actual)]
     OutOfRange {
         path: PathBuf,
-        expected: usize,
-        actual: usize,
+        expected: u64,
+        actual: u64,
     },
 
     #[error("Requested range was invalid")]
@@ -410,7 +404,7 @@ impl ObjectStore for LocalFileSystem {
         let path = self.path_to_filesystem(&location)?;
         maybe_spawn_blocking(move || {
             let (file, metadata) = open_file(&path)?;
-            let meta = convert_metadata(metadata, location)?;
+            let meta = convert_metadata(metadata, location);
             options.check_preconditions(&meta)?;
 
             let range = match options.range {
@@ -430,7 +424,7 @@ impl ObjectStore for LocalFileSystem {
         .await
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let path = self.path_to_filesystem(location)?;
         maybe_spawn_blocking(move || {
             let (mut file, _) = open_file(&path)?;
@@ -439,7 +433,7 @@ impl ObjectStore for LocalFileSystem {
         .await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let path = self.path_to_filesystem(location)?;
         let ranges = ranges.to_vec();
         maybe_spawn_blocking(move || {
@@ -825,7 +819,7 @@ impl Drop for LocalUpload {
 pub(crate) fn chunked_stream(
     mut file: File,
     path: PathBuf,
-    range: Range<usize>,
+    range: Range<u64>,
     chunk_size: usize,
 ) -> BoxStream<'static, Result<Bytes, super::Error>> {
     futures::stream::once(async move {
@@ -847,17 +841,23 @@ pub(crate) fn chunked_stream(
                         return Ok(None);
                     }
 
-                    let to_read = remaining.min(chunk_size);
-                    let mut buffer = Vec::with_capacity(to_read);
+                    let to_read = remaining.min(chunk_size as u64);
+                    let cap = usize::try_from(to_read).map_err(|_e| Error::InvalidRange {
+                        source: InvalidGetRange::TooLarge {
+                            requested: to_read,
+                            max: usize::MAX as u64,
+                        },
+                    })?;
+                    let mut buffer = Vec::with_capacity(cap);
                     let read = (&mut file)
-                        .take(to_read as u64)
+                        .take(to_read)
                         .read_to_end(&mut buffer)
                         .map_err(|e| Error::UnableToReadBytes {
                             source: e,
                             path: path.clone(),
                         })?;
 
-                    Ok(Some((buffer.into(), (file, path, remaining - read))))
+                    Ok(Some((buffer.into(), (file, path, remaining - read as u64))))
                 })
             },
         );
@@ -867,22 +867,18 @@ pub(crate) fn chunked_stream(
     .boxed()
 }
 
-pub(crate) fn read_range(file: &mut File, path: &PathBuf, range: Range<usize>) -> Result<Bytes> {
+pub(crate) fn read_range(file: &mut File, path: &PathBuf, range: Range<u64>) -> Result<Bytes> {
     let to_read = range.end - range.start;
-    file.seek(SeekFrom::Start(range.start as u64))
-        .map_err(|source| {
-            let path = path.into();
-            Error::Seek { source, path }
-        })?;
+    file.seek(SeekFrom::Start(range.start)).map_err(|source| {
+        let path = path.into();
+        Error::Seek { source, path }
+    })?;
 
-    let mut buf = Vec::with_capacity(to_read);
-    let read = file
-        .take(to_read as u64)
-        .read_to_end(&mut buf)
-        .map_err(|source| {
-            let path = path.into();
-            Error::UnableToReadBytes { source, path }
-        })?;
+    let mut buf = Vec::with_capacity(to_read as usize);
+    let read = file.take(to_read).read_to_end(&mut buf).map_err(|source| {
+        let path = path.into();
+        Error::UnableToReadBytes { source, path }
+    })? as u64;
 
     if read != to_read {
         let error = Error::OutOfRange {
@@ -922,7 +918,7 @@ fn open_file(path: &PathBuf) -> Result<(File, Metadata)> {
 
 fn convert_entry(entry: DirEntry, location: Path) -> Result<Option<ObjectMeta>> {
     match entry.metadata() {
-        Ok(metadata) => convert_metadata(metadata, location).map(Some),
+        Ok(metadata) => Ok(Some(convert_metadata(metadata, location))),
         Err(e) => {
             if let Some(io_err) = e.io_error() {
                 if io_err.kind() == ErrorKind::NotFound {
@@ -960,20 +956,16 @@ fn get_etag(metadata: &Metadata) -> String {
     format!("{inode:x}-{mtime:x}-{size:x}")
 }
 
-fn convert_metadata(metadata: Metadata, location: Path) -> Result<ObjectMeta> {
+fn convert_metadata(metadata: Metadata, location: Path) -> ObjectMeta {
     let last_modified = last_modified(&metadata);
-    let size = usize::try_from(metadata.len()).map_err(|source| {
-        let path = location.as_ref().into();
-        Error::FileSizeOverflowedUsize { source, path }
-    })?;
 
-    Ok(ObjectMeta {
+    ObjectMeta {
         location,
         last_modified,
-        size,
+        size: metadata.len(),
         e_tag: Some(get_etag(&metadata)),
         version: None,
-    })
+    }
 }
 
 #[cfg(unix)]
diff --git a/object_store/src/memory.rs b/object_store/src/memory.rs
index 6402f9243..26beff1e1 100644
--- a/object_store/src/memory.rs
+++ b/object_store/src/memory.rs
@@ -241,7 +241,7 @@ impl ObjectStore for InMemory {
         let meta = ObjectMeta {
             location: location.clone(),
             last_modified: entry.last_modified,
-            size: entry.data.len(),
+            size: entry.data.len() as u64,
             e_tag: Some(e_tag),
             version: None,
         };
@@ -250,11 +250,14 @@ impl ObjectStore for InMemory {
         let (range, data) = match options.range {
             Some(range) => {
                 let r = range
-                    .as_range(entry.data.len())
+                    .as_range(entry.data.len() as u64)
                     .map_err(|source| Error::Range { source })?;
-                (r.clone(), entry.data.slice(r))
+                (
+                    r.clone(),
+                    entry.data.slice(r.start as usize..r.end as usize),
+                )
             }
-            None => (0..entry.data.len(), entry.data),
+            None => (0..entry.data.len() as u64, entry.data),
         };
         let stream = futures::stream::once(futures::future::ready(Ok(data)));
 
@@ -266,16 +269,27 @@ impl ObjectStore for InMemory {
         })
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let entry = self.entry(location).await?;
         ranges
             .iter()
             .map(|range| {
                 let r = GetRange::Bounded(range.clone())
-                    .as_range(entry.data.len())
+                    .as_range(entry.data.len() as u64)
                     .map_err(|source| Error::Range { source })?;
-
-                Ok(entry.data.slice(r))
+                let r_end = usize::try_from(r.end).map_err(|_e| Error::Range {
+                    source: InvalidGetRange::TooLarge {
+                        requested: r.end,
+                        max: usize::MAX as u64,
+                    },
+                })?;
+                let r_start = usize::try_from(r.start).map_err(|_e| Error::Range {
+                    source: InvalidGetRange::TooLarge {
+                        requested: r.start,
+                        max: usize::MAX as u64,
+                    },
+                })?;
+                Ok(entry.data.slice(r_start..r_end))
             })
             .collect()
     }
@@ -286,7 +300,7 @@ impl ObjectStore for InMemory {
         Ok(ObjectMeta {
             location: location.clone(),
             last_modified: entry.last_modified,
-            size: entry.data.len(),
+            size: entry.data.len() as u64,
             e_tag: Some(entry.e_tag.to_string()),
             version: None,
         })
@@ -316,7 +330,7 @@ impl ObjectStore for InMemory {
                 Ok(ObjectMeta {
                     location: key.clone(),
                     last_modified: value.last_modified,
-                    size: value.data.len(),
+                    size: value.data.len() as u64,
                     e_tag: Some(value.e_tag.to_string()),
                     version: None,
                 })
@@ -361,7 +375,7 @@ impl ObjectStore for InMemory {
                 let object = ObjectMeta {
                     location: k.clone(),
                     last_modified: v.last_modified,
-                    size: v.data.len(),
+                    size: v.data.len() as u64,
                     e_tag: Some(v.e_tag.to_string()),
                     version: None,
                 };
diff --git a/object_store/src/prefix.rs b/object_store/src/prefix.rs
index a0b67ca4b..ac9803ec0 100644
--- a/object_store/src/prefix.rs
+++ b/object_store/src/prefix.rs
@@ -132,7 +132,7 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
         self.inner.get(&full_path).await
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let full_path = self.full_path(location);
         self.inner.get_range(&full_path, range).await
     }
@@ -142,7 +142,7 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
         self.inner.get_opts(&full_path, options).await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let full_path = self.full_path(location);
         self.inner.get_ranges(&full_path, ranges).await
     }
diff --git a/object_store/src/throttle.rs b/object_store/src/throttle.rs
index 29cd32705..6586ba946 100644
--- a/object_store/src/throttle.rs
+++ b/object_store/src/throttle.rs
@@ -203,7 +203,7 @@ impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
         Ok(throttle_get(result, wait_get_per_byte))
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let config = self.config();
 
         let sleep_duration =
@@ -214,10 +214,10 @@ impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
         self.inner.get_range(location, range).await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let config = self.config();
 
-        let total_bytes: usize = ranges.iter().map(|range| range.end - range.start).sum();
+        let total_bytes: u64 = ranges.iter().map(|range| range.end - range.start).sum();
         let sleep_duration =
             config.wait_get_per_call + config.wait_get_per_byte * total_bytes as u32;
 
diff --git a/object_store/src/util.rs b/object_store/src/util.rs
index 6d638f3cb..17a7a8cad 100644
--- a/object_store/src/util.rs
+++ b/object_store/src/util.rs
@@ -49,7 +49,7 @@ pub(crate) fn hmac_sha256(secret: impl AsRef<[u8]>, bytes: impl AsRef<[u8]>) ->
 }
 
 /// Collect a stream into [`Bytes`] avoiding copying in the event of a single chunk
-pub async fn collect_bytes<S, E>(mut stream: S, size_hint: Option<usize>) -> Result<Bytes, E>
+pub async fn collect_bytes<S, E>(mut stream: S, size_hint: Option<u64>) -> Result<Bytes, E>
 where
     E: Send,
     S: Stream<Item = Result<Bytes, E>> + Send + Unpin,
@@ -60,9 +60,9 @@ where
     match stream.next().await.transpose()? {
         None => Ok(first),
         Some(second) => {
-            let size_hint = size_hint.unwrap_or_else(|| first.len() + second.len());
+            let size_hint = size_hint.unwrap_or_else(|| first.len() as u64 + second.len() as u64);
 
-            let mut buf = Vec::with_capacity(size_hint);
+            let mut buf = Vec::with_capacity(size_hint as usize);
             buf.extend_from_slice(&first);
             buf.extend_from_slice(&second);
             while let Some(maybe_bytes) = stream.next().await {
@@ -89,7 +89,7 @@ where
 
 /// Range requests with a gap less than or equal to this,
 /// will be coalesced into a single request by [`coalesce_ranges`]
-pub const OBJECT_STORE_COALESCE_DEFAULT: usize = 1024 * 1024;
+pub const OBJECT_STORE_COALESCE_DEFAULT: u64 = 1024 * 1024;
 
 /// Up to this number of range requests will be performed in parallel by [`coalesce_ranges`]
 pub(crate) const OBJECT_STORE_COALESCE_PARALLEL: usize = 10;
@@ -103,12 +103,12 @@ pub(crate) const OBJECT_STORE_COALESCE_PARALLEL: usize = 10;
 /// * Make multiple `fetch` requests in parallel (up to maximum of 10)
 ///
 pub async fn coalesce_ranges<F, E, Fut>(
-    ranges: &[Range<usize>],
+    ranges: &[Range<u64>],
     fetch: F,
-    coalesce: usize,
+    coalesce: u64,
 ) -> Result<Vec<Bytes>, E>
 where
-    F: Send + FnMut(Range<usize>) -> Fut,
+    F: Send + FnMut(Range<u64>) -> Fut,
     E: Send,
     Fut: std::future::Future<Output = Result<Bytes, E>> + Send,
 {
@@ -129,13 +129,14 @@ where
 
             let start = range.start - fetch_range.start;
             let end = range.end - fetch_range.start;
-            fetch_bytes.slice(start..end.min(fetch_bytes.len()))
+            let range = (start as usize)..(end as usize).min(fetch_bytes.len());
+            fetch_bytes.slice(range)
         })
         .collect())
 }
 
 /// Returns a sorted list of ranges that cover `ranges`
-fn merge_ranges(ranges: &[Range<usize>], coalesce: usize) -> Vec<Range<usize>> {
+fn merge_ranges(ranges: &[Range<u64>], coalesce: u64) -> Vec<Range<u64>> {
     if ranges.is_empty() {
         return vec![];
     }
@@ -196,38 +197,49 @@ pub enum GetRange {
     /// an error will be returned. Additionally, if the range ends after the end
     /// of the object, the entire remainder of the object will be returned.
     /// Otherwise, the exact requested range will be returned.
-    Bounded(Range<usize>),
+    ///
+    /// Note that range is u64 (i.e., not usize),
+    /// as `object_store` supports 32-bit architectures such as WASM
+    Bounded(Range<u64>),
     /// Request all bytes starting from a given byte offset
-    Offset(usize),
+    Offset(u64),
     /// Request up to the last n bytes
-    Suffix(usize),
+    Suffix(u64),
 }
 
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum InvalidGetRange {
     #[error("Wanted range starting at {requested}, but object was only {length} bytes long")]
-    StartTooLarge { requested: usize, length: usize },
+    StartTooLarge { requested: u64, length: u64 },
 
     #[error("Range started at {start} and ended at {end}")]
-    Inconsistent { start: usize, end: usize },
+    Inconsistent { start: u64, end: u64 },
+
+    #[error("Range {requested} is larger than system memory limit {max}")]
+    TooLarge { requested: u64, max: u64 },
 }
 
 impl GetRange {
     pub(crate) fn is_valid(&self) -> Result<(), InvalidGetRange> {
-        match self {
-            Self::Bounded(r) if r.end <= r.start => {
+        if let Self::Bounded(r) = self {
+            if r.end <= r.start {
                 return Err(InvalidGetRange::Inconsistent {
                     start: r.start,
                     end: r.end,
                 });
             }
-            _ => (),
-        };
+            if (r.end - r.start) > usize::MAX as u64 {
+                return Err(InvalidGetRange::TooLarge {
+                    requested: r.start,
+                    max: usize::MAX as u64,
+                });
+            }
+        }
         Ok(())
     }
 
     /// Convert to a [`Range`] if valid.
-    pub(crate) fn as_range(&self, len: usize) -> Result<Range<usize>, InvalidGetRange> {
+    pub(crate) fn as_range(&self, len: u64) -> Result<Range<u64>, InvalidGetRange> {
         self.is_valid()?;
         match self {
             Self::Bounded(r) => {
@@ -267,7 +279,7 @@ impl Display for GetRange {
     }
 }
 
-impl<T: RangeBounds<usize>> From<T> for GetRange {
+impl<T: RangeBounds<u64>> From<T> for GetRange {
     fn from(value: T) -> Self {
         use std::ops::Bound::*;
         let first = match value.start_bound() {
@@ -323,7 +335,7 @@ mod tests {
     /// Calls coalesce_ranges and validates the returned data is correct
     ///
     /// Returns the fetched ranges
-    async fn do_fetch(ranges: Vec<Range<usize>>, coalesce: usize) -> Vec<Range<usize>> {
+    async fn do_fetch(ranges: Vec<Range<u64>>, coalesce: u64) -> Vec<Range<u64>> {
         let max = ranges.iter().map(|x| x.end).max().unwrap_or(0);
         let src: Vec<_> = (0..max).map(|x| x as u8).collect();
 
@@ -332,7 +344,9 @@ mod tests {
             &ranges,
             |range| {
                 fetches.push(range.clone());
-                futures::future::ready(Ok(Bytes::from(src[range].to_vec())))
+                let start = usize::try_from(range.start).unwrap();
+                let end = usize::try_from(range.end).unwrap();
+                futures::future::ready(Ok(Bytes::from(src[start..end].to_vec())))
             },
             coalesce,
         )
@@ -341,7 +355,10 @@ mod tests {
 
         assert_eq!(ranges.len(), coalesced.len());
         for (range, bytes) in ranges.iter().zip(coalesced) {
-            assert_eq!(bytes.as_ref(), &src[range.clone()]);
+            assert_eq!(
+                bytes.as_ref(),
+                &src[usize::try_from(range.start).unwrap()..usize::try_from(range.end).unwrap()]
+            );
         }
         fetches
     }
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index c14c0e1d3..54992d864 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -69,7 +69,7 @@ paste = { version = "1.0" }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
 sysinfo = { version = "0.33.0", optional = true, default-features = false, features = ["system"] }
 crc32fast = { version = "1.4.2", optional = true, default-features = false }
-simdutf8 = { version = "0.1.5"}
+simdutf8 = { version = "0.1.5", optional = true, default-features = false }
 
 [dev-dependencies]
 base64 = { version = "0.22", default-features = false, features = ["std"] }
@@ -98,7 +98,7 @@ zstd-sys = { version = ">=2.0.0, <2.0.14", default-features = false }
 all-features = true
 
 [features]
-default = ["arrow", "snap", "brotli", "flate2", "lz4", "zstd", "base64"]
+default = ["arrow", "snap", "brotli", "flate2", "lz4", "zstd", "base64", "simdutf8"]
 # Enable lz4
 lz4 = ["lz4_flex"]
 # Enable arrow reader/writer APIs
@@ -121,6 +121,8 @@ zstd = ["dep:zstd", "zstd-sys"]
 sysinfo = ["dep:sysinfo"]
 # Verify 32-bit CRC checksum when decoding parquet pages
 crc = ["dep:crc32fast"]
+# Enable SIMD UTF-8 validation
+simdutf8 = ["dep:simdutf8"]
 
 
 [[example]]
diff --git a/parquet/README.md b/parquet/README.md
index 9ff1d921d..1224e52f3 100644
--- a/parquet/README.md
+++ b/parquet/README.md
@@ -51,17 +51,21 @@ major releases may contain breaking API changes.
 
 The `parquet` crate provides the following features which may be enabled in your `Cargo.toml`:
 
-- `arrow` (default) - support for reading / writing [`arrow`](https://crates.io/crates/arrow) arrays to / from parquet
-- `async` - support `async` APIs for reading parquet
-- `json` - support for reading / writing `json` data to / from parquet
-- `brotli` (default) - support for parquet using `brotli` compression
-- `flate2` (default) - support for parquet using `gzip` compression
-- `lz4` (default) - support for parquet using `lz4` compression
-- `zstd` (default) - support for parquet using `zstd` compression
-- `snap` (default) - support for parquet using `snappy` compression
+- `arrow` (default) - support for reading / writing [`arrow`] arrays to / from Parquet
+- `async` - support `async` APIs for reading Parquet
+- `json` - support for reading / writing `json` data to / from Parquet
+- `brotli` (default) - support for Parquet using `brotli` compression
+- `flate2` (default) - support for Parquet using `gzip` compression
+- `lz4` (default) - support for Parquet using `lz4` compression
+- `zstd` (default) - support for Parquet using `zstd` compression
+- `snap` (default) - support for Parquet using `snappy` compression
 - `cli` - parquet [CLI tools](https://github.com/apache/arrow-rs/tree/main/parquet/src/bin)
 - `crc` - enables functionality to automatically verify checksums of each page (if present) when decoding
 - `experimental` - Experimental APIs which may change, even between minor releases
+- `simdutf8` (default) - Use the [`simdutf8`] crate for SIMD-accelerated UTF-8 validation
+
+[`arrow`]: https://crates.io/crates/arrow
+[`simdutf8`]: https://crates.io/crates/simdutf8
 
 ## Parquet Feature Status
 
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 00627ad61..8df659060 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -27,6 +27,7 @@ use crate::data_type::Int32Type;
 use crate::encodings::decoding::{Decoder, DeltaBitPackDecoder};
 use crate::errors::{ParquetError, Result};
 use crate::schema::types::ColumnDescPtr;
+use crate::util::utf8::check_valid_utf8;
 use arrow_array::{builder::make_view, ArrayRef};
 use arrow_buffer::Buffer;
 use arrow_data::ByteView;
@@ -681,17 +682,6 @@ impl ByteViewArrayDecoderDelta {
     }
 }
 
-/// Check that `val` is a valid UTF-8 sequence
-pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
-    match simdutf8::basic::from_utf8(val) {
-        Ok(_) => Ok(()),
-        Err(_) => {
-            let e = simdutf8::compat::from_utf8(val).unwrap_err();
-            Err(general_err!("encountered non UTF-8 data: {}", e))
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use arrow_array::StringViewArray;
diff --git a/parquet/src/arrow/array_reader/primitive_array.rs b/parquet/src/arrow/array_reader/primitive_array.rs
index a952e00e1..709d0f8bb 100644
--- a/parquet/src/arrow/array_reader/primitive_array.rs
+++ b/parquet/src/arrow/array_reader/primitive_array.rs
@@ -270,6 +270,59 @@ where
 
                 Arc::new(array) as ArrayRef
             }
+            ArrowType::Dictionary(_, value_type) => match value_type.as_ref() {
+                ArrowType::Decimal128(p, s) => {
+                    let array = match array.data_type() {
+                        ArrowType::Int32 => array
+                            .as_any()
+                            .downcast_ref::<Int32Array>()
+                            .unwrap()
+                            .unary(|i| i as i128)
+                            as Decimal128Array,
+                        ArrowType::Int64 => array
+                            .as_any()
+                            .downcast_ref::<Int64Array>()
+                            .unwrap()
+                            .unary(|i| i as i128)
+                            as Decimal128Array,
+                        _ => {
+                            return Err(arrow_err!(
+                                "Cannot convert {:?} to decimal dictionary",
+                                array.data_type()
+                            ));
+                        }
+                    }
+                    .with_precision_and_scale(*p, *s)?;
+
+                    arrow_cast::cast(&array, target_type)?
+                }
+                ArrowType::Decimal256(p, s) => {
+                    let array = match array.data_type() {
+                        ArrowType::Int32 => array
+                            .as_any()
+                            .downcast_ref::<Int32Array>()
+                            .unwrap()
+                            .unary(i256::from)
+                            as Decimal256Array,
+                        ArrowType::Int64 => array
+                            .as_any()
+                            .downcast_ref::<Int64Array>()
+                            .unwrap()
+                            .unary(i256::from)
+                            as Decimal256Array,
+                        _ => {
+                            return Err(arrow_err!(
+                                "Cannot convert {:?} to decimal dictionary",
+                                array.data_type()
+                            ));
+                        }
+                    }
+                    .with_precision_and_scale(*p, *s)?;
+
+                    arrow_cast::cast(&array, target_type)?
+                }
+                _ => arrow_cast::cast(&array, target_type)?,
+            },
             _ => arrow_cast::cast(&array, target_type)?,
         };
 
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 41f15569f..f1081f148 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -843,6 +843,27 @@ fn write_leaf(writer: &mut ColumnWriter<'_>, levels: &ArrayLevels) -> Result<usi
                         .unary::<_, Int32Type>(|v| v.as_i128() as i32);
                     write_primitive(typed, array.values(), levels)
                 }
+                ArrowDataType::Dictionary(_, value_type) => match value_type.as_ref() {
+                    ArrowDataType::Decimal128(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal128Type>()
+                            .unary::<_, Int32Type>(|v| v as i32);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    ArrowDataType::Decimal256(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal256Type>()
+                            .unary::<_, Int32Type>(|v| v.as_i128() as i32);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    _ => {
+                        let array = arrow_cast::cast(column, &ArrowDataType::Int32)?;
+                        let array = array.as_primitive::<Int32Type>();
+                        write_primitive(typed, array.values(), levels)
+                    }
+                },
                 _ => {
                     let array = arrow_cast::cast(column, &ArrowDataType::Int32)?;
                     let array = array.as_primitive::<Int32Type>();
@@ -891,6 +912,27 @@ fn write_leaf(writer: &mut ColumnWriter<'_>, levels: &ArrayLevels) -> Result<usi
                         .unary::<_, Int64Type>(|v| v.as_i128() as i64);
                     write_primitive(typed, array.values(), levels)
                 }
+                ArrowDataType::Dictionary(_, value_type) => match value_type.as_ref() {
+                    ArrowDataType::Decimal128(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal128Type>()
+                            .unary::<_, Int64Type>(|v| v as i64);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    ArrowDataType::Decimal256(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal256Type>()
+                            .unary::<_, Int64Type>(|v| v.as_i128() as i64);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    _ => {
+                        let array = arrow_cast::cast(column, &ArrowDataType::Int64)?;
+                        let array = array.as_primitive::<Int64Type>();
+                        write_primitive(typed, array.values(), levels)
+                    }
+                },
                 _ => {
                     let array = arrow_cast::cast(column, &ArrowDataType::Int64)?;
                     let array = array.as_primitive::<Int64Type>();
@@ -1093,7 +1135,7 @@ mod tests {
     use arrow::util::data_gen::create_random_array;
     use arrow::util::pretty::pretty_format_batches;
     use arrow::{array::*, buffer::Buffer};
-    use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano, NullBuffer};
+    use arrow_buffer::{i256, IntervalDayTime, IntervalMonthDayNano, NullBuffer};
     use arrow_schema::Fields;
     use half::f16;
 
@@ -2670,6 +2712,52 @@ mod tests {
         one_column_roundtrip_with_schema(Arc::new(d), schema);
     }
 
+    #[test]
+    fn arrow_writer_decimal128_dictionary() {
+        let integers = vec![12345, 56789, 34567];
+
+        let keys = UInt8Array::from(vec![Some(0), None, Some(1), Some(2), Some(1)]);
+
+        let values = Decimal128Array::from(integers.clone())
+            .with_precision_and_scale(5, 2)
+            .unwrap();
+
+        let array = DictionaryArray::new(keys, Arc::new(values));
+        one_column_roundtrip(Arc::new(array.clone()), true);
+
+        let values = Decimal128Array::from(integers)
+            .with_precision_and_scale(12, 2)
+            .unwrap();
+
+        let array = array.with_values(Arc::new(values));
+        one_column_roundtrip(Arc::new(array), true);
+    }
+
+    #[test]
+    fn arrow_writer_decimal256_dictionary() {
+        let integers = vec![
+            i256::from_i128(12345),
+            i256::from_i128(56789),
+            i256::from_i128(34567),
+        ];
+
+        let keys = UInt8Array::from(vec![Some(0), None, Some(1), Some(2), Some(1)]);
+
+        let values = Decimal256Array::from(integers.clone())
+            .with_precision_and_scale(5, 2)
+            .unwrap();
+
+        let array = DictionaryArray::new(keys, Arc::new(values));
+        one_column_roundtrip(Arc::new(array.clone()), true);
+
+        let values = Decimal256Array::from(integers)
+            .with_precision_and_scale(12, 2)
+            .unwrap();
+
+        let array = array.with_values(Arc::new(values));
+        one_column_roundtrip(Arc::new(array), true);
+    }
+
     #[test]
     fn arrow_writer_string_dictionary_unsigned_index() {
         // define schema
@@ -3409,4 +3497,34 @@ mod tests {
             "Arrow: Incompatible type. Field 'temperature' has type Float64, array has type Int32"
         );
     }
+
+    #[test]
+    // https://github.com/apache/arrow-rs/issues/6988
+    fn test_roundtrip_empty_schema() {
+        // create empty record batch with empty schema
+        let empty_batch = RecordBatch::try_new_with_options(
+            Arc::new(Schema::empty()),
+            vec![],
+            &RecordBatchOptions::default().with_row_count(Some(0)),
+        )
+        .unwrap();
+
+        // write to parquet
+        let mut parquet_bytes: Vec<u8> = Vec::new();
+        let mut writer =
+            ArrowWriter::try_new(&mut parquet_bytes, empty_batch.schema(), None).unwrap();
+        writer.write(&empty_batch).unwrap();
+        writer.close().unwrap();
+
+        // read from parquet
+        let bytes = Bytes::from(parquet_bytes);
+        let reader = ParquetRecordBatchReaderBuilder::try_new(bytes).unwrap();
+        assert_eq!(reader.schema(), &empty_batch.schema());
+        let batches: Vec<_> = reader
+            .build()
+            .unwrap()
+            .collect::<ArrowResult<Vec<_>>>()
+            .unwrap();
+        assert_eq!(batches.len(), 0);
+    }
 }
diff --git a/parquet/src/arrow/buffer/offset_buffer.rs b/parquet/src/arrow/buffer/offset_buffer.rs
index 8dfb85961..5051dce12 100644
--- a/parquet/src/arrow/buffer/offset_buffer.rs
+++ b/parquet/src/arrow/buffer/offset_buffer.rs
@@ -18,6 +18,7 @@
 use crate::arrow::buffer::bit_util::iter_set_bits_rev;
 use crate::arrow::record_reader::buffer::ValuesBuffer;
 use crate::errors::{ParquetError, Result};
+use crate::util::utf8::check_valid_utf8;
 use arrow_array::{make_array, ArrayRef, OffsetSizeTrait};
 use arrow_buffer::{ArrowNativeType, Buffer};
 use arrow_data::ArrayDataBuilder;
@@ -117,14 +118,7 @@ impl<I: OffsetSizeTrait> OffsetBuffer<I> {
     ///
     /// [`Self::try_push`] can perform this validation check on insertion
     pub fn check_valid_utf8(&self, start_offset: usize) -> Result<()> {
-        match simdutf8::basic::from_utf8(&self.values.as_slice()[start_offset..]) {
-            Ok(_) => Ok(()),
-            Err(_) => {
-                let e = simdutf8::compat::from_utf8(&self.values.as_slice()[start_offset..])
-                    .unwrap_err();
-                Err(general_err!("encountered non UTF-8 data: {}", e))
-            }
-        }
+        check_valid_utf8(&self.values.as_slice()[start_offset..])
     }
 
     /// Converts this into an [`ArrayRef`] with the provided `data_type` and `null_buffer`
diff --git a/parquet/src/lib.rs b/parquet/src/lib.rs
index 3b63845e7..3ca0dbe98 100644
--- a/parquet/src/lib.rs
+++ b/parquet/src/lib.rs
@@ -131,6 +131,9 @@ pub mod data_type;
 pub use self::encodings::{decoding, encoding};
 
 experimental!(#[macro_use] mod util);
+
+pub use util::utf8;
+
 #[cfg(feature = "arrow")]
 pub mod arrow;
 pub mod column;
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index d9e9b22e8..68492e19f 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -1211,6 +1211,13 @@ fn from_thrift_helper(elements: &[SchemaElement], index: usize) -> Result<(usize
         ));
     }
     let element = &elements[index];
+
+    // Check for empty schema
+    if let (true, None | Some(0)) = (is_root_node, element.num_children) {
+        let builder = Type::group_type_builder(&element.name);
+        return Ok((index + 1, Arc::new(builder.build().unwrap())));
+    }
+
     let converted_type = ConvertedType::try_from(element.converted_type)?;
     // LogicalType is only present in v2 Parquet files. ConvertedType is always
     // populated, regardless of the version of the file (v1 or v2).
diff --git a/parquet/src/util/mod.rs b/parquet/src/util/mod.rs
index dfa1285af..143113247 100644
--- a/parquet/src/util/mod.rs
+++ b/parquet/src/util/mod.rs
@@ -19,8 +19,10 @@
 pub mod bit_util;
 mod bit_pack;
 pub(crate) mod interner;
+
 #[cfg(any(test, feature = "test_common"))]
 pub(crate) mod test_common;
+pub mod utf8;
 
 #[cfg(any(test, feature = "test_common"))]
 pub use self::test_common::page_util::{
diff --git a/parquet/src/util/utf8.rs b/parquet/src/util/utf8.rs
new file mode 100644
index 000000000..2a537b5e5
--- /dev/null
+++ b/parquet/src/util/utf8.rs
@@ -0,0 +1,57 @@
+// 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.
+
+//! [`check_valid_utf8`] validation function
+use crate::errors::{ParquetError, Result};
+
+/// Check that `val` is a valid UTF-8 sequence.
+///
+/// If the `simdutf8` feature is enabled, this function will use
+/// SIMD-accelerated validation from the [`simdutf8`] crate. Otherwise, it will use
+/// [`std::str::from_utf8`].
+///
+/// # Errors
+///
+/// Returns `Err::General` with a message compatible with [`std::str::from_utf8`] on failure.
+///
+/// # Example
+/// ```
+/// use parquet::utf8::check_valid_utf8;
+/// assert!(check_valid_utf8(b"hello").is_ok());
+/// assert!(check_valid_utf8(b"hello \xF0\x9F\x98\x8E").is_ok());
+/// // invalid UTF-8
+/// assert!(check_valid_utf8(b"hello \xF0\x9F\x98").is_err());
+/// ```
+///
+/// [`simdutf8`]: https://crates.io/crates/simdutf8
+#[inline(always)]
+pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
+    #[cfg(feature = "simdutf8")]
+    match simdutf8::basic::from_utf8(val) {
+        Ok(_) => Ok(()),
+        Err(_) => {
+            // Use simdutf8::compat to return details about the decoding error
+            let e = simdutf8::compat::from_utf8(val).unwrap_err();
+            Err(general_err!("encountered non UTF-8 data: {}", e))
+        }
+    }
+    #[cfg(not(feature = "simdutf8"))]
+    match std::str::from_utf8(val) {
+        Ok(_) => Ok(()),
+        Err(e) => Err(general_err!("encountered non UTF-8 data: {}", e)),
+    }
+}


Test patch:
diff --git a/object_store/tests/get_range_file.rs b/object_store/tests/get_range_file.rs
index e500fc8ac..6790c11e7 100644
--- a/object_store/tests/get_range_file.rs
+++ b/object_store/tests/get_range_file.rs
@@ -90,12 +90,15 @@ async fn test_get_range() {
     let fetched = store.get(&path).await.unwrap().bytes().await.unwrap();
     assert_eq!(expected, fetched);
 
-    for range in [0..10, 3..5, 0..expected.len()] {
+    for range in [0..10, 3..5, 0..expected.len() as u64] {
         let data = store.get_range(&path, range.clone()).await.unwrap();
-        assert_eq!(&data[..], &expected[range])
+        assert_eq!(
+            &data[..],
+            &expected[range.start as usize..range.end as usize]
+        )
     }
 
-    let over_range = 0..(expected.len() * 2);
+    let over_range = 0..(expected.len() as u64 * 2);
     let data = store.get_range(&path, over_range.clone()).await.unwrap();
     assert_eq!(&data[..], expected)
 }
@@ -113,10 +116,10 @@ async fn test_get_opts_over_range() {
     store.put(&path, expected.clone().into()).await.unwrap();
 
     let opts = GetOptions {
-        range: Some(GetRange::Bounded(0..(expected.len() * 2))),
+        range: Some(GetRange::Bounded(0..(expected.len() as u64 * 2))),
         ..Default::default()
     };
     let res = store.get_opts(&path, opts).await.unwrap();
-    assert_eq!(res.range, 0..expected.len());
+    assert_eq!(res.range, 0..expected.len() as u64);
     assert_eq!(res.bytes().await.unwrap(), expected);
 }
diff --git a/parquet-testing b/parquet-testing
index 4439a223a..f4d7ed772 160000
--- a/parquet-testing
+++ b/parquet-testing
@@ -1,1 +1,1 @@
-Subproject commit 4439a223a315cf874746d3b5da25e6a6b2a2b16e
+Subproject commit f4d7ed772a62a95111db50fbcad2460833e8c882
diff --git a/parquet/tests/arrow_reader/bad_data.rs b/parquet/tests/arrow_reader/bad_data.rs
index cfd61e82d..7de5d7e34 100644
--- a/parquet/tests/arrow_reader/bad_data.rs
+++ b/parquet/tests/arrow_reader/bad_data.rs
@@ -30,6 +30,7 @@ static KNOWN_FILES: &[&str] = &[
     "ARROW-GH-43605.parquet",
     "ARROW-RS-GH-6229-DICTHEADER.parquet",
     "ARROW-RS-GH-6229-LEVELS.parquet",
+    "ARROW-GH-45185.parquet",
     "README.md",
 ];
 
@@ -120,6 +121,16 @@ fn test_arrow_rs_gh_6229_dict_levels() {
     );
 }
 
+#[test]
+#[cfg(feature = "snap")]
+fn test_arrow_rs_gh_45185_dict_levels() {
+    let err = read_file("ARROW-GH-45185.parquet").unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "External: Parquet argument error: Parquet error: first repetition level of batch must be 0"
+    );
+}
+
 /// Reads the file and tries to return the total row count
 /// Returns an error if the file is invalid
 fn read_file(name: &str) -> Result<usize, ParquetError> {