Repository: apache/arrow-rs
Pull request #6671 (2024-11-01 18:30:33)

Support `Utf8View` for `bit_length` kernel
# Which issue does this PR close?

Closes https://github.com/apache/datafusion/issues/13195

# Rationale for this change
 
Thanks to @jayzhan211 , he noticed following issue, array compute kernel `bit_length()` doesn't support `Utf8View` type:
```rust
create table test_source as values
  ('Andrew', 'X'),
  ('Xiangpeng', 'Xiangpeng'),
  ('Raphael', 'R'),
  (NULL, 'R');
create table test as
SELECT
  arrow_cast(column1, 'Utf8View') as column1_utf8view
FROM test_source;

select bit_length(column1_utf8view) from test;
```
caused the error:
```rust
query error DataFusion error: Arrow error: Compute error: bit_length not supported for Utf8View
```

# What changes are included in this PR?

Update `bit_length()` array function to support `Utf8View`

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index 4c967df03..6eae8e246 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -48,7 +48,7 @@ chrono = { workspace = true }
 chrono-tz = { version = "0.10", optional = true }
 num = { version = "0.4.1", default-features = false, features = ["std"] }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
-hashbrown = { version = "0.14.2", default-features = false }
+hashbrown = { version = "0.15.1", default-features = false }
 
 [features]
 ffi = ["arrow-schema/ffi", "arrow-data/ffi"]
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index 5b313913a..9eb598651 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -33,14 +33,18 @@ use std::sync::Arc;
 
 use super::ByteArrayType;
 
-/// [Variable-size Binary View Layout]: An array of variable length bytes view arrays.
+/// [Variable-size Binary View Layout]: An array of variable length bytes views.
+///
+/// This array type is used to store variable length byte data (e.g. Strings, Binary)
+/// and has efficient operations such as `take`, `filter`, and comparison.
 ///
 /// [Variable-size Binary View Layout]: https://arrow.apache.org/docs/format/Columnar.html#variable-size-binary-view-layout
 ///
-/// This is different from [`GenericByteArray`] as it stores both an offset and
-/// length meaning that take / filter operations can be implemented without
-/// copying the underlying data. In addition, it stores an inlined prefix which
-/// can be used to speed up comparisons.
+/// This is different from [`GenericByteArray`], which also stores variable
+/// length byte data, as it represents strings with an offset and length. `take`
+/// and `filter` like operations are implemented by manipulating the "views"
+/// (`u128`) without modifying the bytes. Each view also stores an inlined
+/// prefix which speed up comparisons.
 ///
 /// # See Also
 ///
@@ -50,11 +54,18 @@ use super::ByteArrayType;
 ///
 /// [`ByteView`]: arrow_data::ByteView
 ///
-/// # Notes
+/// # Use the [`eq`] kernel to compare the logical content.
+///
+/// Comparing two `GenericByteViewArray` using PartialEq compares by structure
+/// (the `u128`s) and contents of the buffers, not by logical content. As there
+/// are many different buffer layouts to represent the same data (e.g. different
+/// offsets, different buffer sizes, etc) two arrays with the same data may not
+/// compare equal.
+///
+/// To compare the logical content of two `GenericByteViewArray`s, use the [`eq`]
+/// kernel.
 ///
-/// Comparing two `GenericByteViewArray` using PartialEq compares by structure,
-/// not by value. as there are many different buffer layouts to represent the
-/// same data (e.g. different offsets, different buffer sizes, etc).
+/// [`eq`]: https://docs.rs/arrow/latest/arrow/compute/kernels/cmp/fn.eq.html
 ///
 /// # Layout: "views" and buffers
 ///
@@ -86,6 +97,52 @@ use super::ByteArrayType;
 ///   view and the entire string is stored in one of the buffers. See [`ByteView`]
 ///   to access the fields of the these views.
 ///
+/// As with other arrays, the optimized kernels in [`arrow_compute`] are likely
+/// the easiest and fastest way to work with this data. However, it is possible
+/// to access the views and buffers directly for more control.
+///
+/// For example
+///
+/// ```rust
+/// # use arrow_array::StringViewArray;
+/// # use arrow_array::Array;
+/// use arrow_data::ByteView;
+/// let array = StringViewArray::from(vec![
+///   "hello",
+///   "this string is longer than 12 bytes",
+///   "this string is also longer than 12 bytes"
+/// ]);
+///
+/// // ** Examine the first view (short string) **
+/// assert!(array.is_valid(0)); // Check for nulls
+/// let short_view: u128 = array.views()[0]; // "hello"
+/// // get length of the string
+/// let len = short_view as u32;
+/// assert_eq!(len, 5); // strings less than 12 bytes are stored in the view
+/// // SAFETY: `view` is a valid view
+/// let value = unsafe {
+///   StringViewArray::inline_value(&short_view, len as usize)
+/// };
+/// assert_eq!(value, b"hello");
+///
+/// // ** Examine the third view (long string) **
+/// assert!(array.is_valid(12)); // Check for nulls
+/// let long_view: u128 = array.views()[2]; // "this string is also longer than 12 bytes"
+/// let len = long_view as u32;
+/// assert_eq!(len, 40); // strings longer than 12 bytes are stored in the buffer
+/// let view = ByteView::from(long_view); // use ByteView to access the fields
+/// assert_eq!(view.length, 40);
+/// assert_eq!(view.buffer_index, 0);
+/// assert_eq!(view.offset, 35); // data starts after the first long string
+/// // Views for long strings store a 4 byte prefix
+/// let prefix = view.prefix.to_le_bytes();
+/// assert_eq!(&prefix, b"this");
+/// let value = array.value(2); // get the string value (see `value` implementation for how to access the bytes directly)
+/// assert_eq!(value, "this string is also longer than 12 bytes");
+/// ```
+///
+/// [`arrow_compute`]: https://docs.rs/arrow/latest/arrow/compute/index.html
+///
 /// Unlike [`GenericByteArray`], there are no constraints on the offsets other
 /// than they must point into a valid buffer. However, they can be out of order,
 /// non continuous and overlapping.
@@ -694,6 +751,8 @@ where
 
 /// A [`GenericByteViewArray`] of `[u8]`
 ///
+/// See [`GenericByteViewArray`] for format and layout details.
+///
 /// # Example
 /// ```
 /// use arrow_array::BinaryViewArray;
@@ -733,6 +792,8 @@ impl From<Vec<Option<&[u8]>>> for BinaryViewArray {
 
 /// A [`GenericByteViewArray`] that stores utf8 data
 ///
+/// See [`GenericByteViewArray`] for format and layout details.
+///
 /// # Example
 /// ```
 /// use arrow_array::StringViewArray;
diff --git a/arrow-json/src/writer/encoder.rs b/arrow-json/src/writer/encoder.rs
index 84ed384cf..ed430fe6a 100644
--- a/arrow-json/src/writer/encoder.rs
+++ b/arrow-json/src/writer/encoder.rs
@@ -454,7 +454,7 @@ impl Encoder for ArrayFormatter<'_> {
 /// A newtype wrapper around [`ArrayFormatter`] that skips surrounding the value with `"`
 struct RawArrayFormatter<'a>(ArrayFormatter<'a>);
 
-impl<'a> Encoder for RawArrayFormatter<'a> {
+impl Encoder for RawArrayFormatter<'_> {
     fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
         let _ = write!(out, "{}", self.0.value(idx));
     }
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index d616b01e7..d96dad2f1 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -436,18 +436,21 @@ where
 
     let mut start = 0i64;
     let mut i = 0;
-    let filter_values = pred.filter.values();
     let mut count = R::default_value();
+    let filter_values = pred.filter.values();
 
     for end in run_ends.inner().into_iter().map(|i| (*i).into()) {
         let mut keep = false;
-        // in filter_array the predicate array is checked to have the same len as the run end array
-        // this means the largest value in the run_ends is == to pred.len()
-        // so we're always within bounds when calling value_unchecked
-        for pred in (start..end).map(|i| unsafe { filter_values.value_unchecked(i as usize) }) {
+
+        for pred in filter_values
+            .iter()
+            .skip(start as usize)
+            .take((end - start) as usize)
+        {
             count += R::Native::from(pred);
             keep |= pred
         }
+
         // this is to avoid branching
         new_run_ends[i] = count;
         i += keep as usize;
@@ -1280,6 +1283,26 @@ mod tests {
         assert_eq!(0, actual.len());
     }
 
+    #[test]
+    fn test_filter_run_end_encoding_array_max_value_gt_predicate_len() {
+        let run_ends = Int64Array::from(vec![2, 3, 8, 10]);
+        let values = Int64Array::from(vec![7, -2, 9, -8]);
+        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
+        let b = BooleanArray::from(vec![false, true, true]);
+        let c = filter(&a, &b).unwrap();
+        let actual: &RunArray<Int64Type> = as_run_array(&c);
+        assert_eq!(2, actual.len());
+
+        let expected = RunArray::try_new(
+            &Int64Array::from(vec![1, 2]),
+            &Int64Array::from(vec![7, -2]),
+        )
+        .expect("Failed to make expected RunArray test is broken");
+
+        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
+        assert_eq!(actual.values(), expected.values())
+    }
+
     #[test]
     fn test_filter_dictionary_array() {
         let values = [Some("hello"), None, Some("world"), Some("!")];
diff --git a/arrow-string/src/length.rs b/arrow-string/src/length.rs
index 97f876a9f..6a28d44ea 100644
--- a/arrow-string/src/length.rs
+++ b/arrow-string/src/length.rs
@@ -137,6 +137,15 @@ pub fn bit_length(array: &dyn Array) -> Result<ArrayRef, ArrowError> {
             let list = array.as_string::<i64>();
             Ok(bit_length_impl::<Int64Type>(list.offsets(), list.nulls()))
         }
+        DataType::Utf8View => {
+            let list = array.as_string_view();
+            let values = list
+                .views()
+                .iter()
+                .map(|view| (*view as i32).wrapping_mul(8))
+                .collect();
+            Ok(Arc::new(Int32Array::new(values, array.nulls().cloned())))
+        }
         DataType::Binary => {
             let list = array.as_binary::<i32>();
             Ok(bit_length_impl::<Int32Type>(list.offsets(), list.nulls()))
@@ -462,6 +471,35 @@ mod tests {
             })
     }
 
+    #[test]
+    fn bit_length_test_utf8view() {
+        bit_length_cases()
+            .into_iter()
+            .for_each(|(input, len, expected)| {
+                let string_array = StringViewArray::from(input);
+                let result = bit_length(&string_array).unwrap();
+                assert_eq!(len, result.len());
+                let result = result.as_any().downcast_ref::<Int32Array>().unwrap();
+                expected.iter().enumerate().for_each(|(i, value)| {
+                    assert_eq!(*value, result.value(i));
+                });
+            })
+    }
+
+    #[test]
+    fn bit_length_null_utf8view() {
+        bit_length_null_cases()
+            .into_iter()
+            .for_each(|(input, len, expected)| {
+                let array = StringArray::from(input);
+                let result = bit_length(&array).unwrap();
+                assert_eq!(len, result.len());
+                let result = result.as_any().downcast_ref::<Int32Array>().unwrap();
+
+                let expected: Int32Array = expected.into();
+                assert_eq!(&expected, result);
+            })
+    }
     #[test]
     fn bit_length_binary() {
         let value: Vec<&[u8]> = vec![b"one", &[0xff, 0xf8], b"three"];
diff --git a/arrow-string/src/predicate.rs b/arrow-string/src/predicate.rs
index f559088e6..408d9d45c 100644
--- a/arrow-string/src/predicate.rs
+++ b/arrow-string/src/predicate.rs
@@ -239,7 +239,7 @@ fn equals_kernel((n, h): (&u8, &u8)) -> bool {
 }
 
 fn equals_ignore_ascii_case_kernel((n, h): (&u8, &u8)) -> bool {
-    n.to_ascii_lowercase() == h.to_ascii_lowercase()
+    n.eq_ignore_ascii_case(h)
 }
 
 /// Transforms a like `pattern` to a regex compatible pattern. To achieve that, it does:
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index cab2ac31c..86d1392eb 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -46,7 +46,7 @@ walkdir = "2"
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
-quick-xml = { version = "0.36.0", features = ["serialize", "overlapped-lists"], optional = true }
+quick-xml = { version = "0.37.0", features = ["serialize", "overlapped-lists"], optional = true }
 serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
 serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index a259f6bd9..ccc9c341f 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -79,6 +79,9 @@ enum Error {
     #[snafu(display("Error getting put response body: {}", source))]
     PutResponseBody { source: reqwest::Error },
 
+    #[snafu(display("Got invalid put request: {}", source))]
+    InvalidPutRequest { source: quick_xml::se::SeError },
+
     #[snafu(display("Got invalid put response: {}", source))]
     InvalidPutResponse { source: quick_xml::de::DeError },
 
@@ -495,7 +498,7 @@ impl GoogleCloudStorageClient {
         let credential = self.get_credential().await?;
 
         let data = quick_xml::se::to_string(&upload_info)
-            .context(InvalidPutResponseSnafu)?
+            .context(InvalidPutRequestSnafu)?
             // We cannot disable the escaping that transforms "/" to "&quote;" :(
             // https://github.com/tafia/quick-xml/issues/362
             // https://github.com/tafia/quick-xml/issues/350
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 32bc13b62..4064baba0 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -63,7 +63,7 @@ serde_json = { version = "1.0", default-features = false, features = ["std"], op
 seq-macro = { version = "0.3", default-features = false }
 futures = { version = "0.3", default-features = false, features = ["std"], optional = true }
 tokio = { version = "1.0", optional = true, default-features = false, features = ["macros", "rt", "io-util"] }
-hashbrown = { version = "0.14", default-features = false }
+hashbrown = { version = "0.15", default-features = false }
 twox-hash = { version = "1.6", default-features = false }
 paste = { version = "1.0" }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
@@ -81,7 +81,7 @@ lz4_flex = { version = "0.11", default-features = false, features = ["std", "fra
 zstd = { version = "0.13", default-features = false }
 serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
-tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
+tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
 object_store = { version = "0.11.0", default-features = false, features = ["azure"] }
 
diff --git a/parquet/src/arrow/async_reader/store.rs b/parquet/src/arrow/async_reader/store.rs
index e6b47856e..fd0397b5e 100644
--- a/parquet/src/arrow/async_reader/store.rs
+++ b/parquet/src/arrow/async_reader/store.rs
@@ -15,17 +15,15 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::ops::Range;
-use std::sync::Arc;
+use std::{ops::Range, sync::Arc};
 
 use bytes::Bytes;
-use futures::future::BoxFuture;
-use futures::{FutureExt, TryFutureExt};
-
-use object_store::{ObjectMeta, ObjectStore};
+use futures::{future::BoxFuture, FutureExt, TryFutureExt};
+use object_store::{path::Path, ObjectMeta, ObjectStore};
+use tokio::runtime::Handle;
 
 use crate::arrow::async_reader::AsyncFileReader;
-use crate::errors::Result;
+use crate::errors::{ParquetError, Result};
 use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
 
 /// Reads Parquet files in object storage using [`ObjectStore`].
@@ -59,6 +57,7 @@ pub struct ParquetObjectReader {
     metadata_size_hint: Option<usize>,
     preload_column_index: bool,
     preload_offset_index: bool,
+    runtime: Option<Handle>,
 }
 
 impl ParquetObjectReader {
@@ -72,6 +71,7 @@ impl ParquetObjectReader {
             metadata_size_hint: None,
             preload_column_index: false,
             preload_offset_index: false,
+            runtime: None,
         }
     }
 
@@ -99,29 +99,70 @@ impl ParquetObjectReader {
             ..self
         }
     }
+
+    /// Perform IO on the provided tokio runtime
+    ///
+    /// Tokio is a cooperative scheduler, and relies on tasks yielding in a timely manner
+    /// to service IO. Therefore, running IO and CPU-bound tasks, such as parquet decoding,
+    /// on the same tokio runtime can lead to degraded throughput, dropped connections and
+    /// other issues. For more information see [here].
+    ///
+    /// [here]: https://www.influxdata.com/blog/using-rustlangs-async-tokio-runtime-for-cpu-bound-tasks/
+    pub fn with_runtime(self, handle: Handle) -> Self {
+        Self {
+            runtime: Some(handle),
+            ..self
+        }
+    }
+
+    fn spawn<F, O, E>(&self, f: F) -> BoxFuture<'_, Result<O>>
+    where
+        F: for<'a> FnOnce(&'a Arc<dyn ObjectStore>, &'a Path) -> BoxFuture<'a, Result<O, E>>
+            + Send
+            + 'static,
+        O: Send + 'static,
+        E: Into<ParquetError> + Send + 'static,
+    {
+        match &self.runtime {
+            Some(handle) => {
+                let path = self.meta.location.clone();
+                let store = Arc::clone(&self.store);
+                handle
+                    .spawn(async move { f(&store, &path).await })
+                    .map_ok_or_else(
+                        |e| match e.try_into_panic() {
+                            Err(e) => Err(ParquetError::External(Box::new(e))),
+                            Ok(p) => std::panic::resume_unwind(p),
+                        },
+                        |res| res.map_err(|e| e.into()),
+                    )
+                    .boxed()
+            }
+            None => f(&self.store, &self.meta.location)
+                .map_err(|e| e.into())
+                .boxed(),
+        }
+    }
 }
 
 impl AsyncFileReader for ParquetObjectReader {
     fn get_bytes(&mut self, range: Range<usize>) -> BoxFuture<'_, Result<Bytes>> {
-        self.store
-            .get_range(&self.meta.location, range)
-            .map_err(|e| e.into())
-            .boxed()
+        self.spawn(|store, path| store.get_range(path, range))
     }
 
     fn get_byte_ranges(&mut self, ranges: Vec<Range<usize>>) -> BoxFuture<'_, Result<Vec<Bytes>>>
     where
         Self: Send,
     {
-        async move {
-            self.store
-                .get_ranges(&self.meta.location, &ranges)
-                .await
-                .map_err(|e| e.into())
-        }
-        .boxed()
+        self.spawn(|store, path| async move { store.get_ranges(path, &ranges).await }.boxed())
     }
 
+    // This method doesn't directly call `self.spawn` because all of the IO that is done down the
+    // line due to this method call is done through `self.get_bytes` and/or `self.get_byte_ranges`.
+    // When `self` is passed into `ParquetMetaDataReader::load_and_finish`, it treats it as
+    // an `impl MetadataFetch` and calls those methods to get data from it. Due to `Self`'s impl of
+    // `AsyncFileReader`, the calls to `MetadataFetch::fetch` are just delegated to
+    // `Self::get_bytes`.
     fn get_metadata(&mut self) -> BoxFuture<'_, Result<Arc<ParquetMetaData>>> {
         Box::pin(async move {
             let file_size = self.meta.size;
@@ -138,30 +179,40 @@ impl AsyncFileReader for ParquetObjectReader {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    };
 
     use futures::TryStreamExt;
 
     use arrow::util::test_util::parquet_test_data;
+    use futures::FutureExt;
     use object_store::local::LocalFileSystem;
     use object_store::path::Path;
-    use object_store::ObjectStore;
+    use object_store::{ObjectMeta, ObjectStore};
 
-    use crate::arrow::async_reader::ParquetObjectReader;
+    use crate::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
     use crate::arrow::ParquetRecordBatchStreamBuilder;
+    use crate::errors::ParquetError;
 
-    #[tokio::test]
-    async fn test_simple() {
+    async fn get_meta_store() -> (ObjectMeta, Arc<dyn ObjectStore>) {
         let res = parquet_test_data();
         let store = LocalFileSystem::new_with_prefix(res).unwrap();
 
-        let mut meta = store
+        let meta = store
             .head(&Path::from("alltypes_plain.parquet"))
             .await
             .unwrap();
 
-        let store = Arc::new(store) as Arc<dyn ObjectStore>;
-        let object_reader = ParquetObjectReader::new(Arc::clone(&store), meta.clone());
+        (meta, Arc::new(store) as Arc<dyn ObjectStore>)
+    }
+
+    #[tokio::test]
+    async fn test_simple() {
+        let (meta, store) = get_meta_store().await;
+        let object_reader = ParquetObjectReader::new(store, meta);
+
         let builder = ParquetRecordBatchStreamBuilder::new(object_reader)
             .await
             .unwrap();
@@ -169,7 +220,11 @@ mod tests {
 
         assert_eq!(batches.len(), 1);
         assert_eq!(batches[0].num_rows(), 8);
+    }
 
+    #[tokio::test]
+    async fn test_not_found() {
+        let (mut meta, store) = get_meta_store().await;
         meta.location = Path::from("I don't exist.parquet");
 
         let object_reader = ParquetObjectReader::new(store, meta);
@@ -180,10 +235,86 @@ mod tests {
                 let err = e.to_string();
                 assert!(
                     err.contains("not found: No such file or directory (os error 2)"),
-                    "{}",
-                    err
+                    "{err}",
                 );
             }
         }
     }
+
+    #[tokio::test]
+    async fn test_runtime_is_used() {
+        let num_actions = Arc::new(AtomicUsize::new(0));
+
+        let (a1, a2) = (num_actions.clone(), num_actions.clone());
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .on_thread_park(move || {
+                a1.fetch_add(1, Ordering::Relaxed);
+            })
+            .on_thread_unpark(move || {
+                a2.fetch_add(1, Ordering::Relaxed);
+            })
+            .build()
+            .unwrap();
+
+        let (meta, store) = get_meta_store().await;
+
+        let initial_actions = num_actions.load(Ordering::Relaxed);
+
+        let reader = ParquetObjectReader::new(store, meta).with_runtime(rt.handle().clone());
+
+        let builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap();
+        let batches: Vec<_> = builder.build().unwrap().try_collect().await.unwrap();
+
+        // Just copied these assert_eqs from the `test_simple` above
+        assert_eq!(batches.len(), 1);
+        assert_eq!(batches[0].num_rows(), 8);
+
+        assert!(num_actions.load(Ordering::Relaxed) - initial_actions > 0);
+
+        // Runtimes have to be dropped in blocking contexts, so we need to move this one to a new
+        // blocking thread to drop it.
+        tokio::runtime::Handle::current().spawn_blocking(move || drop(rt));
+    }
+
+    /// Unit test that `ParquetObjectReader::spawn`spawns on the provided runtime
+    #[tokio::test]
+    async fn test_runtime_thread_id_different() {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .build()
+            .unwrap();
+
+        let (meta, store) = get_meta_store().await;
+
+        let reader = ParquetObjectReader::new(store, meta).with_runtime(rt.handle().clone());
+
+        let current_id = std::thread::current().id();
+
+        let other_id = reader
+            .spawn(|_, _| async move { Ok::<_, ParquetError>(std::thread::current().id()) }.boxed())
+            .await
+            .unwrap();
+
+        assert_ne!(current_id, other_id);
+
+        tokio::runtime::Handle::current().spawn_blocking(move || drop(rt));
+    }
+
+    #[tokio::test]
+    async fn io_fails_on_shutdown_runtime() {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .build()
+            .unwrap();
+
+        let (meta, store) = get_meta_store().await;
+
+        let mut reader = ParquetObjectReader::new(store, meta).with_runtime(rt.handle().clone());
+
+        rt.shutdown_background();
+
+        let err = reader.get_bytes(0..1).await.unwrap_err().to_string();
+
+        assert!(err.to_string().contains("was cancelled"));
+    }
 }
diff --git a/parquet/src/errors.rs b/parquet/src/errors.rs
index bb4d2543c..6adbffa2a 100644
--- a/parquet/src/errors.rs
+++ b/parquet/src/errors.rs
@@ -106,7 +106,6 @@ impl From<str::Utf8Error> for ParquetError {
         ParquetError::External(Box::new(e))
     }
 }
-
 #[cfg(feature = "arrow")]
 impl From<ArrowError> for ParquetError {
     fn from(e: ArrowError) -> ParquetError {
diff --git a/parquet/src/record/reader.rs b/parquet/src/record/reader.rs
index 1f9128a8b..fd6ca7cdd 100644
--- a/parquet/src/record/reader.rs
+++ b/parquet/src/record/reader.rs
@@ -138,7 +138,17 @@ impl TreeBuilder {
                 .column_descr_ptr();
             let col_reader = row_group_reader.get_column_reader(orig_index)?;
             let column = TripletIter::new(col_descr, col_reader, self.batch_size);
-            Reader::PrimitiveReader(field, Box::new(column))
+            let reader = Reader::PrimitiveReader(field.clone(), Box::new(column));
+            if repetition == Repetition::REPEATED {
+                Reader::RepeatedReader(
+                    field,
+                    curr_def_level - 1,
+                    curr_rep_level - 1,
+                    Box::new(reader),
+                )
+            } else {
+                reader
+            }
         } else {
             match field.get_basic_info().converted_type() {
                 // List types
@@ -1688,6 +1698,131 @@ mod tests {
         assert_eq!(rows, expected_rows);
     }
 
+    #[test]
+    fn test_tree_reader_handle_primitive_repeated_fields_with_no_annotation() {
+        // In this test the REPEATED fields are primitives
+        let rows = test_file_reader_rows("repeated_primitive_no_list.parquet", None).unwrap();
+        let expected_rows = vec![
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list([0, 1, 2, 3].map(Field::Int).to_vec()))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["foo", "zero", "one", "two"]
+                            .map(|s| Field::Str(s.to_string()))
+                            .to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list([0, 1, 2, 3].map(Field::Int).to_vec()))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["foo", "zero", "one", "two"]
+                                    .map(|s| Field::Str(s.to_string()))
+                                    .to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list(vec![]))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["three"].map(|s| Field::Str(s.to_string())).to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list(vec![]))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["three"].map(|s| Field::Str(s.to_string())).to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list(vec![Field::Int(4)]))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["four"].map(|s| Field::Str(s.to_string())).to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list(vec![Field::Int(4)]))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["four"].map(|s| Field::Str(s.to_string())).to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list([5, 6, 7, 8].map(Field::Int).to_vec()))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["five", "six", "seven", "eight"]
+                            .map(|s| Field::Str(s.to_string()))
+                            .to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list([5, 6, 7, 8].map(Field::Int).to_vec()))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["five", "six", "seven", "eight"]
+                                    .map(|s| Field::Str(s.to_string()))
+                                    .to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+        ];
+        assert_eq!(rows, expected_rows);
+    }
+
     fn test_file_reader_rows(file_name: &str, schema: Option<Type>) -> Result<Vec<Row>> {
         let file = get_test_file(file_name);
         let file_reader: Box<dyn FileReader> = Box::new(SerializedFileReader::new(file)?);


Test patch:
diff --git a/parquet-testing b/parquet-testing
index 50af3d8ce..550368ca7 160000
--- a/parquet-testing
+++ b/parquet-testing
@@ -1,1 +1,1 @@
-Subproject commit 50af3d8ce206990d81014b1862e5ce7380dc3e08
+Subproject commit 550368ca77b97231efead39251a96bd6f8f08c6e