Repository: apache/arrow-rs
Pull request #6987 (2025-01-16 05:18:46)

fix: Parquet column writer `Dictionary(_, Decimal128)` and `Dictionary(_, Decimal256)`
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes #.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

Currently Parquet reads/writes for Arrow `Dictionary(Decimal)` type where Decimal is INT-based is falling back to `arrow_cast::cast`, which leads to losing precision.

# 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.
-->

For `Dictionary(Decimal)` type where decimal is int-based, writer and reader now perform `arrow_cast::cast` only for dictionary packing/unpacking, while values are only casted to required integer type without being modified, similar to how regular `Decimal` works.

# Are there any user-facing changes?
No

<!--
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 call them out.
-->



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-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/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> {