Repository: jorgecarleitao/arrow2
Pull request #213 (2021-07-22 20:34:18)

Added FFI for remaining types
* Duration
* Timestamp
* Decimal
* FixedSizedBinary
* FixedSizedList


codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#213](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (6ea63c3) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/8d7d08fc6a8343bded84d596196857433df40eda?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (8d7d08f) will **decrease** coverage by `0.06%`.
> The diff coverage is `38.29%`.

[![Impacted file tree graph](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213/graphs/tree.svg?width=650&height=150&src=pr&token=AgyTF60R3D&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)

```diff
@@            Coverage Diff             @@
##             main     #213      +/-   ##
==========================================
- Coverage   76.81%   76.75%   -0.07%     
==========================================
  Files         226      226              
  Lines       19446    19490      +44     
==========================================
+ Hits        14938    14959      +21     
- Misses       4508     4531      +23     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/array/fixed\_size\_list/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2ZpeGVkX3NpemVfbGlzdC9tb2QucnM=) | `58.33% <0.00%> (-2.54%)` | :arrow_down: |
| [src/ffi/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9mZmkucnM=) | `71.38% <34.14%> (-3.71%)` | :arrow_down: |
| [src/ffi/array.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9hcnJheS5ycw==) | `80.82% <100.00%> (+1.11%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [8d7d08f...6ea63c3](https://codecov.io/gh/jorgecarleitao/arrow2/pull/213?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).


Patch:
diff --git a/README.md b/README.md
index c9c71f9d68..d40f2baff1 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,9 @@
 # Arrow2: Transmute-free Arrow
 
-![test](https://github.com/jorgecarleitao/arrow2/actions/workflows/test.yml/badge.svg)
+[![test](https://github.com/jorgecarleitao/arrow2/actions/workflows/test.yml/badge.svg)](https://github.com/jorgecarleitao/arrow2/actions/workflows/Build.yml)
+[![](https://img.shields.io/crates/d/arrow2.svg)](https://crates.io/crates/arrow2)
+[![](https://img.shields.io/crates/dv/arrow2.svg)](https://crates.io/crates/arrow2)
+[![](https://docs.rs/arrow2/badge.svg)](https://docs.rs/arrow2/)
 
 This repository contains a Rust library to work with the [Arrow format](https://arrow.apache.org/).
 It is a re-write of the [official Arrow crate](https://github.com/apache/arrow) using
@@ -35,6 +38,7 @@ To not run them, pass `ARROW2_IGNORE_PARQUET` to the tests (the tests will be ma
 ```bash
 git clone git@github.com:jorgecarleitao/arrow2.git
 cd arrow2
+git submodule update --init
 ARROW2_IGNORE_PARQUET= cargo test
 ```
 
diff --git a/benches/sort_kernel.rs b/benches/sort_kernel.rs
index a55a6971fa..ed48c4c8e4 100644
--- a/benches/sort_kernel.rs
+++ b/benches/sort_kernel.rs
@@ -35,11 +35,11 @@ fn bench_lexsort(arr_a: &dyn Array, array_b: &dyn Array) {
         },
     ];
 
-    criterion::black_box(lexsort(&columns).unwrap());
+    criterion::black_box(lexsort(&columns, None).unwrap());
 }
 
 fn bench_sort(arr_a: &dyn Array) {
-    sort(criterion::black_box(arr_a), &SortOptions::default()).unwrap();
+    sort(criterion::black_box(arr_a), &SortOptions::default(), None).unwrap();
 }
 
 fn add_benchmark(c: &mut Criterion) {
@@ -66,6 +66,11 @@ fn add_benchmark(c: &mut Criterion) {
         c.bench_function(&format!("lexsort null 2^{} f32", log2_size), |b| {
             b.iter(|| bench_lexsort(&arr_a, &arr_b))
         });
+
+        let arr_a = create_string_array::<i32>(size, 0.1);
+        c.bench_function(&format!("sort utf8 null 2^{}", log2_size), |b| {
+            b.iter(|| bench_sort(&arr_a))
+        });
     });
 }
 
diff --git a/benches/take_kernels.rs b/benches/take_kernels.rs
index 7b2c32749b..2ab746b443 100644
--- a/benches/take_kernels.rs
+++ b/benches/take_kernels.rs
@@ -46,7 +46,7 @@ fn create_random_index(size: usize, null_density: f32) -> PrimitiveArray<i32> {
 }
 
 fn bench_take(values: &dyn Array, indices: &PrimitiveArray<i32>) {
-    criterion::black_box(take::take(values, &indices).unwrap());
+    criterion::black_box(take::take(values, indices).unwrap());
 }
 
 fn add_benchmark(c: &mut Criterion) {
diff --git a/examples/csv_read.rs b/examples/csv_read.rs
index 2f7a66c7c1..2999f0d8b5 100644
--- a/examples/csv_read.rs
+++ b/examples/csv_read.rs
@@ -37,7 +37,7 @@ fn main() -> Result<()> {
 
     let file_path = &args[1];
 
-    let batch = read_path(&file_path, None)?;
+    let batch = read_path(file_path, None)?;
     println!("{:?}", batch);
     Ok(())
 }
diff --git a/examples/parquet_read.rs b/examples/parquet_read.rs
index dceacdaf6a..e3b82b4714 100644
--- a/examples/parquet_read.rs
+++ b/examples/parquet_read.rs
@@ -45,7 +45,7 @@ fn main() -> Result<()> {
     let column = args[2].parse::<usize>().unwrap();
     let row_group = args[3].parse::<usize>().unwrap();
 
-    let array = read_column_chunk(&file_path, row_group, column)?;
+    let array = read_column_chunk(file_path, row_group, column)?;
     println!("{}", array);
     Ok(())
 }
diff --git a/src/array/binary/ffi.rs b/src/array/binary/ffi.rs
index e93d4f93a4..2771ee9a27 100644
--- a/src/array/binary/ffi.rs
+++ b/src/array/binary/ffi.rs
@@ -10,17 +10,11 @@ use super::BinaryArray;
 
 unsafe impl<O: Offset> ToFfi for BinaryArray<O> {
     fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
-        unsafe {
-            vec![
-                self.validity.as_ref().map(|x| x.as_ptr()),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.offsets.as_ptr() as *mut u8
-                )),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.values.as_ptr() as *mut u8
-                )),
-            ]
-        }
+        vec![
+            self.validity.as_ref().map(|x| x.as_ptr()),
+            std::ptr::NonNull::new(self.offsets.as_ptr() as *mut u8),
+            std::ptr::NonNull::new(self.values.as_ptr() as *mut u8),
+        ]
     }
 
     #[inline]
diff --git a/src/array/binary/mod.rs b/src/array/binary/mod.rs
index 11d79878ee..96425bac9e 100644
--- a/src/array/binary/mod.rs
+++ b/src/array/binary/mod.rs
@@ -168,9 +168,9 @@ mod tests {
             array.validity(),
             &Some(Bitmap::from_u8_slice(&[0b00000101], 3))
         );
-        assert_eq!(array.is_valid(0), true);
-        assert_eq!(array.is_valid(1), false);
-        assert_eq!(array.is_valid(2), true);
+        assert!(array.is_valid(0));
+        assert!(!array.is_valid(1));
+        assert!(array.is_valid(2));
 
         let array2 = BinaryArray::<i32>::from_data(
             array.offsets().clone(),
diff --git a/src/array/boolean/mod.rs b/src/array/boolean/mod.rs
index d0c95df0c1..af5087ee93 100644
--- a/src/array/boolean/mod.rs
+++ b/src/array/boolean/mod.rs
@@ -141,24 +141,24 @@ mod tests {
 
         let array: BooleanArray = data.into_iter().collect();
 
-        assert_eq!(array.value(0), true);
-        assert_eq!(array.value(1), false);
-        assert_eq!(array.value(2), false);
+        assert!(array.value(0));
+        assert!(!array.value(1));
+        assert!(!array.value(2));
         assert_eq!(array.values(), &Bitmap::from_u8_slice(&[0b00000001], 3));
         assert_eq!(
             array.validity(),
             &Some(Bitmap::from_u8_slice(&[0b00000101], 3))
         );
-        assert_eq!(array.is_valid(0), true);
-        assert_eq!(array.is_valid(1), false);
-        assert_eq!(array.is_valid(2), true);
+        assert!(array.is_valid(0));
+        assert!(!array.is_valid(1));
+        assert!(array.is_valid(2));
 
         let array2 = BooleanArray::from_data(array.values().clone(), array.validity().clone());
         assert_eq!(array, array2);
 
         let array = array.slice(1, 2);
-        assert_eq!(array.value(0), false);
-        assert_eq!(array.value(1), false);
+        assert!(!array.value(0));
+        assert!(!array.value(1));
     }
 
     #[test]
diff --git a/src/array/equal/mod.rs b/src/array/equal/mod.rs
index 5dfdbf812b..45062e27c4 100644
--- a/src/array/equal/mod.rs
+++ b/src/array/equal/mod.rs
@@ -424,8 +424,8 @@ mod tests {
 
     pub(super) fn test_equal(lhs: &dyn Array, rhs: &dyn Array, expected: bool) {
         // equality is symmetric
-        assert_eq!(equal(lhs, lhs), true, "\n{:?}\n{:?}", lhs, lhs);
-        assert_eq!(equal(rhs, rhs), true, "\n{:?}\n{:?}", rhs, rhs);
+        assert!(equal(lhs, lhs), "\n{:?}\n{:?}", lhs, lhs);
+        assert!(equal(rhs, rhs), "\n{:?}\n{:?}", rhs, rhs);
 
         assert_eq!(equal(lhs, rhs), expected, "\n{:?}\n{:?}", lhs, rhs);
         assert_eq!(equal(rhs, lhs), expected, "\n{:?}\n{:?}", rhs, lhs);
diff --git a/src/array/fixed_size_binary/mod.rs b/src/array/fixed_size_binary/mod.rs
index 1da1d24d36..ee4405cfd2 100644
--- a/src/array/fixed_size_binary/mod.rs
+++ b/src/array/fixed_size_binary/mod.rs
@@ -133,14 +133,10 @@ impl std::fmt::Display for FixedSizeBinaryArray {
 
 unsafe impl ToFfi for FixedSizeBinaryArray {
     fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
-        unsafe {
-            vec![
-                self.validity.as_ref().map(|x| x.as_ptr()),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.values.as_ptr() as *mut u8
-                )),
-            ]
-        }
+        vec![
+            self.validity.as_ref().map(|x| x.as_ptr()),
+            std::ptr::NonNull::new(self.values.as_ptr() as *mut u8),
+        ]
     }
 
     fn offset(&self) -> usize {
diff --git a/src/array/fixed_size_list/mod.rs b/src/array/fixed_size_list/mod.rs
index 2b5c392303..df5877162d 100644
--- a/src/array/fixed_size_list/mod.rs
+++ b/src/array/fixed_size_list/mod.rs
@@ -133,4 +133,8 @@ unsafe impl ToFfi for FixedSizeListArray {
     fn offset(&self) -> usize {
         self.offset
     }
+
+    fn children(&self) -> Vec<Arc<dyn Array>> {
+        vec![self.values().clone()]
+    }
 }
diff --git a/src/array/growable/dictionary.rs b/src/array/growable/dictionary.rs
index e67aa72bf9..94667b8785 100644
--- a/src/array/growable/dictionary.rs
+++ b/src/array/growable/dictionary.rs
@@ -25,7 +25,7 @@ fn concatenate_values<K: DictionaryKey>(
     arrays_values: &[&dyn Array],
     capacity: usize,
 ) -> (Arc<dyn Array>, Vec<usize>) {
-    let mut mutable = make_growable(&arrays_values, false, capacity);
+    let mut mutable = make_growable(arrays_values, false, capacity);
     let mut offsets = Vec::with_capacity(arrays_keys.len() + 1);
     offsets.push(0);
     for (i, values) in arrays_values.iter().enumerate() {
diff --git a/src/array/list/ffi.rs b/src/array/list/ffi.rs
index ebc256b7d0..cd794ff5cc 100644
--- a/src/array/list/ffi.rs
+++ b/src/array/list/ffi.rs
@@ -7,14 +7,10 @@ use super::ListArray;
 
 unsafe impl<O: Offset> ToFfi for ListArray<O> {
     fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
-        unsafe {
-            vec![
-                self.validity.as_ref().map(|x| x.as_ptr()),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.offsets.as_ptr() as *mut u8
-                )),
-            ]
-        }
+        vec![
+            self.validity.as_ref().map(|x| x.as_ptr()),
+            std::ptr::NonNull::new(self.offsets.as_ptr() as *mut u8),
+        ]
     }
 
     fn offset(&self) -> usize {
diff --git a/src/array/ord.rs b/src/array/ord.rs
index 2529ea423d..8157dfbf72 100644
--- a/src/array/ord.rs
+++ b/src/array/ord.rs
@@ -92,7 +92,7 @@ fn compare_f64<'a>(left: &'a dyn Array, right: &'a dyn Array) -> DynComparator<'
 fn compare_string<'a, O: Offset>(left: &'a dyn Array, right: &'a dyn Array) -> DynComparator<'a> {
     let left = left.as_any().downcast_ref::<Utf8Array<O>>().unwrap();
     let right = right.as_any().downcast_ref::<Utf8Array<O>>().unwrap();
-    Box::new(move |i, j| left.value(i).cmp(&right.value(j)))
+    Box::new(move |i, j| left.value(i).cmp(right.value(j)))
 }
 
 fn compare_dict<'a, K>(
diff --git a/src/array/primitive/ffi.rs b/src/array/primitive/ffi.rs
index 68871899bb..87980219c0 100644
--- a/src/array/primitive/ffi.rs
+++ b/src/array/primitive/ffi.rs
@@ -10,14 +10,10 @@ use super::PrimitiveArray;
 
 unsafe impl<T: NativeType> ToFfi for PrimitiveArray<T> {
     fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
-        unsafe {
-            vec![
-                self.validity.as_ref().map(|x| x.as_ptr()),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.values.as_ptr() as *mut u8
-                )),
-            ]
-        }
+        vec![
+            self.validity.as_ref().map(|x| x.as_ptr()),
+            std::ptr::NonNull::new(self.values.as_ptr() as *mut u8),
+        ]
     }
 
     #[inline]
diff --git a/src/array/primitive/mod.rs b/src/array/primitive/mod.rs
index c299957a84..5a394e817c 100644
--- a/src/array/primitive/mod.rs
+++ b/src/array/primitive/mod.rs
@@ -225,9 +225,9 @@ mod tests {
             array.validity(),
             &Some(Bitmap::from_u8_slice(&[0b00000101], 3))
         );
-        assert_eq!(array.is_valid(0), true);
-        assert_eq!(array.is_valid(1), false);
-        assert_eq!(array.is_valid(2), true);
+        assert!(array.is_valid(0));
+        assert!(!array.is_valid(1));
+        assert!(array.is_valid(2));
 
         let array2 = PrimitiveArray::<i32>::from_data(
             DataType::Int32,
diff --git a/src/array/specification.rs b/src/array/specification.rs
index 5acfc9e1b5..70576cb862 100644
--- a/src/array/specification.rs
+++ b/src/array/specification.rs
@@ -8,8 +8,9 @@ use crate::{
 };
 
 /// Trait describing any type that can be used to index a slot of an array.
-pub trait Index: NativeType {
+pub trait Index: NativeType + NaturalDataType {
     fn to_usize(&self) -> usize;
+    fn from_usize(index: usize) -> Option<Self>;
 }
 
 /// Trait describing types that can be used as offsets as per Arrow specification.
@@ -17,14 +18,12 @@ pub trait Index: NativeType {
 /// # Safety
 /// Do not implement.
 pub unsafe trait Offset:
-    Index + NaturalDataType + Num + Ord + std::ops::AddAssign + std::ops::Sub + num::CheckedAdd
+    Index + Num + Ord + std::ops::AddAssign + std::ops::Sub + num::CheckedAdd
 {
     fn is_large() -> bool;
 
     fn to_isize(&self) -> isize;
 
-    fn from_usize(value: usize) -> Option<Self>;
-
     fn from_isize(value: isize) -> Option<Self>;
 }
 
@@ -34,11 +33,6 @@ unsafe impl Offset for i32 {
         false
     }
 
-    #[inline]
-    fn from_usize(value: usize) -> Option<Self> {
-        Self::try_from(value).ok()
-    }
-
     #[inline]
     fn from_isize(value: isize) -> Option<Self> {
         Self::try_from(value).ok()
@@ -56,11 +50,6 @@ unsafe impl Offset for i64 {
         true
     }
 
-    #[inline]
-    fn from_usize(value: usize) -> Option<Self> {
-        Some(value as i64)
-    }
-
     #[inline]
     fn from_isize(value: isize) -> Option<Self> {
         Self::try_from(value).ok()
@@ -77,6 +66,11 @@ impl Index for i32 {
     fn to_usize(&self) -> usize {
         *self as usize
     }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
 }
 
 impl Index for i64 {
@@ -84,6 +78,11 @@ impl Index for i64 {
     fn to_usize(&self) -> usize {
         *self as usize
     }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
 }
 
 impl Index for u32 {
@@ -91,6 +90,11 @@ impl Index for u32 {
     fn to_usize(&self) -> usize {
         *self as usize
     }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
 }
 
 impl Index for u64 {
@@ -98,6 +102,11 @@ impl Index for u64 {
     fn to_usize(&self) -> usize {
         *self as usize
     }
+
+    #[inline]
+    fn from_usize(value: usize) -> Option<Self> {
+        Self::try_from(value).ok()
+    }
 }
 
 #[inline]
diff --git a/src/array/utf8/ffi.rs b/src/array/utf8/ffi.rs
index dcf3db2d60..adc2a1ea86 100644
--- a/src/array/utf8/ffi.rs
+++ b/src/array/utf8/ffi.rs
@@ -8,17 +8,11 @@ use super::Utf8Array;
 
 unsafe impl<O: Offset> ToFfi for Utf8Array<O> {
     fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
-        unsafe {
-            vec![
-                self.validity.as_ref().map(|x| x.as_ptr()),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.offsets.as_ptr() as *mut u8
-                )),
-                Some(std::ptr::NonNull::new_unchecked(
-                    self.values.as_ptr() as *mut u8
-                )),
-            ]
-        }
+        vec![
+            self.validity.as_ref().map(|x| x.as_ptr()),
+            std::ptr::NonNull::new(self.offsets.as_ptr() as *mut u8),
+            std::ptr::NonNull::new(self.values.as_ptr() as *mut u8),
+        ]
     }
 
     fn offset(&self) -> usize {
diff --git a/src/array/utf8/mod.rs b/src/array/utf8/mod.rs
index 314b0c4e15..11f6ebdd10 100644
--- a/src/array/utf8/mod.rs
+++ b/src/array/utf8/mod.rs
@@ -222,9 +222,9 @@ mod tests {
             array.validity(),
             &Some(Bitmap::from_u8_slice(&[0b00000101], 3))
         );
-        assert_eq!(array.is_valid(0), true);
-        assert_eq!(array.is_valid(1), false);
-        assert_eq!(array.is_valid(2), true);
+        assert!(array.is_valid(0));
+        assert!(!array.is_valid(1));
+        assert!(array.is_valid(2));
 
         let array2 = Utf8Array::<i32>::from_data(
             array.offsets().clone(),
diff --git a/src/bitmap/bitmap_ops.rs b/src/bitmap/bitmap_ops.rs
index 2b491c2e67..67b036ff00 100644
--- a/src/bitmap/bitmap_ops.rs
+++ b/src/bitmap/bitmap_ops.rs
@@ -174,7 +174,7 @@ impl Not for &Bitmap {
     type Output = Bitmap;
 
     fn not(self) -> Bitmap {
-        unary(&self, |a| !a)
+        unary(self, |a| !a)
     }
 }
 
@@ -194,28 +194,28 @@ mod test {
     fn test_eq() {
         let lhs = create_bitmap([0b01101010], 8);
         let rhs = create_bitmap([0b01001110], 8);
-        assert_eq!(eq(&lhs, &rhs), false);
-        assert_eq!(eq(&lhs, &lhs), true);
+        assert!(!eq(&lhs, &rhs));
+        assert!(eq(&lhs, &lhs));
     }
 
     #[test]
     fn test_eq_len() {
         let lhs = create_bitmap([0b01101010], 6);
         let rhs = create_bitmap([0b00101010], 6);
-        assert_eq!(eq(&lhs, &rhs), true);
+        assert!(eq(&lhs, &rhs));
         let rhs = create_bitmap([0b00001010], 6);
-        assert_eq!(eq(&lhs, &rhs), false);
+        assert!(!eq(&lhs, &rhs));
     }
 
     #[test]
     fn test_eq_slice() {
         let lhs = create_bitmap([0b10101010], 8).slice(1, 7);
         let rhs = create_bitmap([0b10101011], 8).slice(1, 7);
-        assert_eq!(eq(&lhs, &rhs), true);
+        assert!(eq(&lhs, &rhs));
 
         let lhs = create_bitmap([0b10101010], 8).slice(2, 6);
         let rhs = create_bitmap([0b10101110], 8).slice(2, 6);
-        assert_eq!(eq(&lhs, &rhs), false);
+        assert!(!eq(&lhs, &rhs));
     }
 
     #[test]
diff --git a/src/bitmap/mutable.rs b/src/bitmap/mutable.rs
index 509d8f6cec..f2b94e2709 100644
--- a/src/bitmap/mutable.rs
+++ b/src/bitmap/mutable.rs
@@ -560,7 +560,7 @@ mod tests {
         unsafe { b.extend_from_trusted_len_iter_unchecked(iter) };
         let b: Bitmap = b.into();
         let mut iter = b.iter().enumerate();
-        assert_eq!(iter.next().unwrap().1, true);
+        assert!(iter.next().unwrap().1);
         for (i, v) in iter {
             assert_eq!((i - 1) % 6 == 0, v);
         }
@@ -570,14 +570,14 @@ mod tests {
     fn test_set() {
         let mut bitmap = MutableBitmap::from_len_zeroed(12);
         bitmap.set(0, true);
-        assert_eq!(bitmap.get(0), true);
+        assert!(bitmap.get(0));
         bitmap.set(0, false);
-        assert_eq!(bitmap.get(0), false);
+        assert!(!bitmap.get(0));
 
         bitmap.set(11, true);
-        assert_eq!(bitmap.get(11), true);
+        assert!(bitmap.get(11));
         bitmap.set(11, false);
-        assert_eq!(bitmap.get(11), false);
+        assert!(!bitmap.get(11));
         bitmap.set(11, true);
 
         let bitmap: Option<Bitmap> = bitmap.into();
diff --git a/src/bitmap/utils/mod.rs b/src/bitmap/utils/mod.rs
index ab43b25d82..db284b28d7 100644
--- a/src/bitmap/utils/mod.rs
+++ b/src/bitmap/utils/mod.rs
@@ -91,21 +91,21 @@ mod tests {
             0b01000000, 0b11111111,
         ];
         for i in 0..8 {
-            assert_eq!(get_bit(input, i), false);
+            assert!(!get_bit(input, i));
         }
-        assert_eq!(get_bit(input, 8), true);
+        assert!(get_bit(input, 8));
         for i in 8 + 1..2 * 8 {
-            assert_eq!(get_bit(input, i), false);
+            assert!(!get_bit(input, i));
         }
-        assert_eq!(get_bit(input, 2 * 8 + 1), true);
+        assert!(get_bit(input, 2 * 8 + 1));
         for i in 2 * 8 + 2..3 * 8 {
-            assert_eq!(get_bit(input, i), false);
+            assert!(!get_bit(input, i));
         }
-        assert_eq!(get_bit(input, 3 * 8 + 2), true);
+        assert!(get_bit(input, 3 * 8 + 2));
         for i in 3 * 8 + 3..4 * 8 {
-            assert_eq!(get_bit(input, i), false);
+            assert!(!get_bit(input, i));
         }
-        assert_eq!(get_bit(input, 4 * 8 + 3), true);
+        assert!(get_bit(input, 4 * 8 + 3));
     }
 
     #[test]
diff --git a/src/buffer/immutable.rs b/src/buffer/immutable.rs
index 0a7ef6a6e0..065bfa47ba 100644
--- a/src/buffer/immutable.rs
+++ b/src/buffer/immutable.rs
@@ -171,14 +171,14 @@ mod tests {
     fn test_new() {
         let buffer = Buffer::<i32>::new();
         assert_eq!(buffer.len(), 0);
-        assert_eq!(buffer.is_empty(), true);
+        assert!(buffer.is_empty());
     }
 
     #[test]
     fn test_new_zeroed() {
         let buffer = Buffer::<i32>::new_zeroed(2);
         assert_eq!(buffer.len(), 2);
-        assert_eq!(buffer.is_empty(), false);
+        assert!(!buffer.is_empty());
         assert_eq!(buffer.as_slice(), &[0, 0]);
     }
 
diff --git a/src/buffer/mutable.rs b/src/buffer/mutable.rs
index 911bf7f18f..7cf5cdae39 100644
--- a/src/buffer/mutable.rs
+++ b/src/buffer/mutable.rs
@@ -186,6 +186,15 @@ impl<T: NativeType> MutableBuffer<T> {
         self.len = 0
     }
 
+    /// Shortens the buffer.
+    /// If `len` is greater or equal to the buffers' current length, this has no effect.
+    #[inline]
+    pub fn truncate(&mut self, len: usize) {
+        if len < self.len {
+            self.len = len;
+        }
+    }
+
     /// Returns the data stored in this buffer as a slice.
     #[inline]
     pub fn as_slice(&self) -> &[T] {
@@ -363,6 +372,8 @@ impl<T: NativeType> MutableBuffer<T> {
     /// # Safety
     /// This method assumes that the iterator's size is correct and is undefined behavior
     /// to use it on an iterator that reports an incorrect length.
+    // This inline has been validated to offer 50% improvement in operations like `take`.
+    #[inline]
     pub unsafe fn extend_from_trusted_len_iter_unchecked<I: Iterator<Item = T>>(
         &mut self,
         iterator: I,
@@ -452,6 +463,8 @@ impl<T: NativeType> MutableBuffer<T> {
     /// # Safety
     /// This method assumes that the iterator's size is correct and is undefined behavior
     /// to use it on an iterator that reports an incorrect length.
+    // This inline has been validated to offer 50% improvement in operations like `take`.
+    #[inline]
     pub unsafe fn try_from_trusted_len_iter_unchecked<
         E,
         I: Iterator<Item = std::result::Result<T, E>>,
@@ -621,21 +634,21 @@ mod tests {
     fn default() {
         let b = MutableBuffer::<i32>::default();
         assert_eq!(b.len(), 0);
-        assert_eq!(b.is_empty(), true);
+        assert!(b.is_empty());
     }
 
     #[test]
     fn with_capacity() {
         let b = MutableBuffer::<i32>::with_capacity(6);
         assert!(b.capacity() >= 6);
-        assert_eq!(b.is_empty(), true);
+        assert!(b.is_empty());
     }
 
     #[test]
     fn from_len_zeroed() {
         let b = MutableBuffer::<i32>::from_len_zeroed(3);
         assert_eq!(b.len(), 3);
-        assert_eq!(b.is_empty(), false);
+        assert!(!b.is_empty());
         assert_eq!(b.as_slice(), &[0, 0, 0]);
     }
 
diff --git a/src/compute/aggregate/min_max.rs b/src/compute/aggregate/min_max.rs
index b47573781a..fdc08f05a3 100644
--- a/src/compute/aggregate/min_max.rs
+++ b/src/compute/aggregate/min_max.rs
@@ -34,7 +34,7 @@ fn min_max_string<O: Offset, F: Fn(&str, &str) -> bool>(
 
         for i in 0..array.len() {
             let item = array.value(i);
-            if validity.get_bit(i) && (!has_value || cmp(&n, item)) {
+            if validity.get_bit(i) && (!has_value || cmp(n, item)) {
                 has_value = true;
                 n = item;
             }
@@ -45,7 +45,7 @@ fn min_max_string<O: Offset, F: Fn(&str, &str) -> bool>(
         for i in 1..array.len() {
             // loop is up to `len`.
             let item = unsafe { array.value_unchecked(i) };
-            if cmp(&n, item) {
+            if cmp(n, item) {
                 n = item;
             }
         }
diff --git a/src/compute/arithmetics/time.rs b/src/compute/arithmetics/time.rs
index 4bdb5354ef..509fbc4a38 100644
--- a/src/compute/arithmetics/time.rs
+++ b/src/compute/arithmetics/time.rs
@@ -496,7 +496,7 @@ mod tests {
         let expected = PrimitiveArray::from(&vec![Some(10i64), Some(20i64), None, Some(30i64)])
             .to(DataType::Duration(TimeUnit::Second));
 
-        let result = subtract_timestamps(&timestamp_a, &&timestamp_b).unwrap();
+        let result = subtract_timestamps(&timestamp_a, &timestamp_b).unwrap();
         assert_eq!(result, expected);
     }
 
@@ -526,7 +526,7 @@ mod tests {
         ])
         .to(DataType::Duration(TimeUnit::Millisecond));
 
-        let result = subtract_timestamps(&timestamp_a, &&timestamp_b).unwrap();
+        let result = subtract_timestamps(&timestamp_a, &timestamp_b).unwrap();
         assert_eq!(result, expected);
     }
 
diff --git a/src/compute/arity.rs b/src/compute/arity.rs
index 2cd56d9a88..0e99c488e1 100644
--- a/src/compute/arity.rs
+++ b/src/compute/arity.rs
@@ -120,7 +120,7 @@ where
     // the iteration, then the validity is changed to None to mark the value
     // as Null
     let bitmap: Bitmap = mut_bitmap.into();
-    let validity = combine_validities(&array.validity(), &Some(bitmap));
+    let validity = combine_validities(array.validity(), &Some(bitmap));
 
     PrimitiveArray::<O>::from_data(data_type, values, validity)
 }
diff --git a/src/compute/boolean.rs b/src/compute/boolean.rs
index 1ce6032e72..9fe1d0aac6 100644
--- a/src/compute/boolean.rs
+++ b/src/compute/boolean.rs
@@ -37,7 +37,7 @@ where
     let left_buffer = lhs.values();
     let right_buffer = rhs.values();
 
-    let values = op(&left_buffer, &right_buffer);
+    let values = op(left_buffer, right_buffer);
 
     Ok(BooleanArray::from_data(values, validity))
 }
@@ -60,7 +60,7 @@ where
 /// # }
 /// ```
 pub fn and(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
-    binary_boolean_kernel(&lhs, &rhs, |lhs, rhs| lhs & rhs)
+    binary_boolean_kernel(lhs, rhs, |lhs, rhs| lhs & rhs)
 }
 
 /// Performs `OR` operation on two arrays. If either left or right value is null then the
@@ -81,7 +81,7 @@ pub fn and(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
 /// # }
 /// ```
 pub fn or(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
-    binary_boolean_kernel(&lhs, &rhs, |lhs, rhs| lhs | rhs)
+    binary_boolean_kernel(lhs, rhs, |lhs, rhs| lhs | rhs)
 }
 
 /// Performs unary `NOT` operation on an arrays. If value is null then the result is also
@@ -352,7 +352,7 @@ mod tests {
         let b = b.slice(2, 4);
         let b = b.as_any().downcast_ref::<BooleanArray>().unwrap();
 
-        let c = and(&a, &b).unwrap();
+        let c = and(a, b).unwrap();
 
         let expected = BooleanArray::from(vec![Some(false), Some(false), None, Some(true)]);
 
diff --git a/src/compute/boolean_kleene.rs b/src/compute/boolean_kleene.rs
index 788a6344f9..1cdd2b41c7 100644
--- a/src/compute/boolean_kleene.rs
+++ b/src/compute/boolean_kleene.rs
@@ -37,8 +37,8 @@ pub fn or(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
     let validity = match (lhs_validity, rhs_validity) {
         (Some(lhs_validity), Some(rhs_validity)) => {
             Some(quaternary(
-                &lhs_values,
-                &rhs_values,
+                lhs_values,
+                rhs_values,
                 lhs_validity,
                 rhs_validity,
                 // see https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics
@@ -55,8 +55,8 @@ pub fn or(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
         (Some(lhs_validity), None) => {
             // B != U
             Some(ternary(
-                &lhs_values,
-                &rhs_values,
+                lhs_values,
+                rhs_values,
                 lhs_validity,
                 // see https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics
                 |lhs, rhs, lhs_v| {
@@ -71,8 +71,8 @@ pub fn or(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
         }
         (None, Some(rhs_validity)) => {
             Some(ternary(
-                &lhs_values,
-                &rhs_values,
+                lhs_values,
+                rhs_values,
                 rhs_validity,
                 // see https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics
                 |lhs, rhs, rhs_v| {
@@ -123,8 +123,8 @@ pub fn and(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
     let validity = match (lhs_validity, rhs_validity) {
         (Some(lhs_validity), Some(rhs_validity)) => {
             Some(quaternary(
-                &lhs_values,
-                &rhs_values,
+                lhs_values,
+                rhs_values,
                 lhs_validity,
                 rhs_validity,
                 // see https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics
@@ -140,8 +140,8 @@ pub fn and(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
         }
         (Some(lhs_validity), None) => {
             Some(ternary(
-                &lhs_values,
-                &rhs_values,
+                lhs_values,
+                rhs_values,
                 lhs_validity,
                 // see https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics
                 |lhs, rhs, lhs_v| {
@@ -156,8 +156,8 @@ pub fn and(lhs: &BooleanArray, rhs: &BooleanArray) -> Result<BooleanArray> {
         }
         (None, Some(rhs_validity)) => {
             Some(ternary(
-                &lhs_values,
-                &rhs_values,
+                lhs_values,
+                rhs_values,
                 rhs_validity,
                 // see https://en.wikipedia.org/wiki/Three-valued_logic#Kleene_and_Priest_logics
                 |lhs, rhs, rhs_v| {
diff --git a/src/compute/cast/primitive_to.rs b/src/compute/cast/primitive_to.rs
index 8041d49885..dd34d93461 100644
--- a/src/compute/cast/primitive_to.rs
+++ b/src/compute/cast/primitive_to.rs
@@ -205,8 +205,8 @@ pub fn time64_to_time32(
     from_unit: &TimeUnit,
     to_unit: &TimeUnit,
 ) -> PrimitiveArray<i32> {
-    let from_size = time_unit_multiple(&from_unit);
-    let to_size = time_unit_multiple(&to_unit);
+    let from_size = time_unit_multiple(from_unit);
+    let to_size = time_unit_multiple(to_unit);
     let divisor = from_size / to_size;
     unary(
         from,
@@ -221,8 +221,8 @@ pub fn timestamp_to_timestamp(
     to_unit: &TimeUnit,
     tz: &Option<String>,
 ) -> PrimitiveArray<i64> {
-    let from_size = time_unit_multiple(&from_unit);
-    let to_size = time_unit_multiple(&to_unit);
+    let from_size = time_unit_multiple(from_unit);
+    let to_size = time_unit_multiple(to_unit);
     let to_type = DataType::Timestamp(to_unit.clone(), tz.clone());
     // we either divide or multiply, depending on size of each unit
     if from_size >= to_size {
diff --git a/src/compute/cast/timestamps.rs b/src/compute/cast/timestamps.rs
index 79335fee51..354dd652a8 100644
--- a/src/compute/cast/timestamps.rs
+++ b/src/compute/cast/timestamps.rs
@@ -208,7 +208,7 @@ mod tests {
         // Note: Use chrono APIs that are different than
         // naive_datetime_to_timestamp to compute the utc offset to
         // try and double check the logic
-        let utc_offset_secs = match Local.offset_from_local_datetime(&naive_datetime) {
+        let utc_offset_secs = match Local.offset_from_local_datetime(naive_datetime) {
             LocalResult::Single(local_offset) => local_offset.fix().local_minus_utc() as i64,
             _ => panic!("Unexpected failure converting to local datetime"),
         };
diff --git a/src/compute/comparison/mod.rs b/src/compute/comparison/mod.rs
index d7d1857f3f..8a8697e6a2 100644
--- a/src/compute/comparison/mod.rs
+++ b/src/compute/comparison/mod.rs
@@ -236,11 +236,10 @@ mod tests {
 
         datatypes.into_iter().for_each(|d1| {
             let array = new_null_array(d1.clone(), 10);
+            let op = Operator::Eq;
             if can_compare(&d1) {
-                let op = Operator::Eq;
                 assert!(compare(array.as_ref(), array.as_ref(), op).is_ok());
             } else {
-                let op = Operator::Eq;
                 assert!(compare(array.as_ref(), array.as_ref(), op).is_err());
             }
         });
diff --git a/src/compute/filter.rs b/src/compute/filter.rs
index d69d0b9b9c..cc3d837c7a 100644
--- a/src/compute/filter.rs
+++ b/src/compute/filter.rs
@@ -316,9 +316,9 @@ mod tests {
         assert_eq!(67, d.len());
         assert_eq!(3, d.null_count());
         assert_eq!(1, d.value(0));
-        assert_eq!(true, d.is_null(1));
+        assert!(d.is_null(1));
         assert_eq!(64, d.value(63));
-        assert_eq!(true, d.is_null(64));
+        assert!(d.is_null(64));
         assert_eq!(67, d.value(65));
     }
 
@@ -344,7 +344,7 @@ mod tests {
         let c = filter(&a, &b).unwrap();
         let d = c.as_ref().as_any().downcast_ref::<Int32Array>().unwrap();
         assert_eq!(1, d.len());
-        assert_eq!(true, d.is_null(0));
+        assert!(d.is_null(0));
     }
 
     #[test]
@@ -359,8 +359,8 @@ mod tests {
             .unwrap();
         assert_eq!(2, d.len());
         assert_eq!("hello", d.value(0));
-        assert_eq!(false, d.is_null(0));
-        assert_eq!(true, d.is_null(1));
+        assert!(!d.is_null(0));
+        assert!(d.is_null(1));
     }
 
     #[test]
@@ -376,8 +376,8 @@ mod tests {
             .unwrap();
         assert_eq!(2, d.len());
         assert_eq!(b"hello", d.value(0));
-        assert_eq!(false, d.is_null(0));
-        assert_eq!(true, d.is_null(1));
+        assert!(!d.is_null(0));
+        assert!(d.is_null(1));
     }
 
     /*
diff --git a/src/compute/merge_sort/mod.rs b/src/compute/merge_sort/mod.rs
index 339d07aca3..5c4fddfaac 100644
--- a/src/compute/merge_sort/mod.rs
+++ b/src/compute/merge_sort/mod.rs
@@ -637,8 +637,8 @@ mod tests {
         let options = SortOptions::default();
 
         // sort individually, potentially in parallel.
-        let a0 = sort(a0, &options)?;
-        let a1 = sort(a1, &options)?;
+        let a0 = sort(a0, &options, None)?;
+        let a1 = sort(a1, &options, None)?;
 
         // merge then. If multiple arrays, this can be applied in parallel.
         let result = merge_sort(a0.as_ref(), a1.as_ref(), &options)?;
diff --git a/src/compute/regex_match.rs b/src/compute/regex_match.rs
index f8e2f9f2b5..3e8d8220b6 100644
--- a/src/compute/regex_match.rs
+++ b/src/compute/regex_match.rs
@@ -104,7 +104,7 @@ mod tests {
     ) {
         let lhs = Utf8Array::<O>::from_slice(lhs);
         let expected = BooleanArray::from_slice(expected);
-        let result = op(&lhs, &pattern).unwrap();
+        let result = op(&lhs, pattern).unwrap();
         assert_eq!(result, expected);
     }
 
diff --git a/src/compute/sort/boolean.rs b/src/compute/sort/boolean.rs
new file mode 100644
index 0000000000..2e7082a590
--- /dev/null
+++ b/src/compute/sort/boolean.rs
@@ -0,0 +1,51 @@
+use crate::{
+    array::{Array, BooleanArray, Index, PrimitiveArray},
+    buffer::MutableBuffer,
+};
+
+use super::SortOptions;
+
+/// Returns the indices that would sort a [`BooleanArray`].
+pub fn sort_boolean<I: Index>(
+    values: &BooleanArray,
+    value_indices: Vec<I>,
+    null_indices: Vec<I>,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> PrimitiveArray<I> {
+    let descending = options.descending;
+
+    // create tuples that are used for sorting
+    let mut valids = value_indices
+        .into_iter()
+        .map(|index| (index, values.value(index.to_usize())))
+        .collect::<Vec<(I, bool)>>();
+
+    let mut nulls = null_indices;
+
+    if !descending {
+        valids.sort_by(|a, b| a.1.cmp(&b.1));
+    } else {
+        valids.sort_by(|a, b| a.1.cmp(&b.1).reverse());
+        // reverse to keep a stable ordering
+        nulls.reverse();
+    }
+
+    let mut values = MutableBuffer::<I>::with_capacity(values.len());
+
+    if options.nulls_first {
+        values.extend_from_slice(nulls.as_slice());
+        valids.iter().for_each(|x| values.push(x.0));
+    } else {
+        // nulls last
+        valids.iter().for_each(|x| values.push(x.0));
+        values.extend_from_slice(nulls.as_slice());
+    }
+
+    // un-efficient; there are much more performant ways of sorting nulls above, anyways.
+    if let Some(limit) = limit {
+        values.truncate(limit);
+    }
+
+    PrimitiveArray::<I>::from_data(I::DATA_TYPE, values.into(), None)
+}
diff --git a/src/compute/sort/common.rs b/src/compute/sort/common.rs
new file mode 100644
index 0000000000..c36763831b
--- /dev/null
+++ b/src/compute/sort/common.rs
@@ -0,0 +1,185 @@
+use crate::{
+    array::{Index, PrimitiveArray},
+    bitmap::Bitmap,
+    buffer::MutableBuffer,
+};
+
+use super::SortOptions;
+
+/// # Safety
+/// This function guarantees that:
+/// * `get` is only called for `0 <= i < limit`
+/// * `cmp` is only called from the co-domain of `get`.
+#[inline]
+fn k_element_sort_inner<I: Index, T, G, F>(
+    indices: &mut [I],
+    get: G,
+    descending: bool,
+    limit: usize,
+    mut cmp: F,
+) where
+    G: Fn(usize) -> T,
+    F: FnMut(&T, &T) -> std::cmp::Ordering,
+{
+    if descending {
+        let compare = |lhs: &I, rhs: &I| {
+            let lhs = get(lhs.to_usize());
+            let rhs = get(rhs.to_usize());
+            cmp(&lhs, &rhs).reverse()
+        };
+        let (before, _, _) = indices.select_nth_unstable_by(limit, compare);
+        let compare = |lhs: &I, rhs: &I| {
+            let lhs = get(lhs.to_usize());
+            let rhs = get(rhs.to_usize());
+            cmp(&lhs, &rhs).reverse()
+        };
+        before.sort_unstable_by(compare);
+    } else {
+        let compare = |lhs: &I, rhs: &I| {
+            let lhs = get(lhs.to_usize());
+            let rhs = get(rhs.to_usize());
+            cmp(&lhs, &rhs)
+        };
+        let (before, _, _) = indices.select_nth_unstable_by(limit, compare);
+        let compare = |lhs: &I, rhs: &I| {
+            let lhs = get(lhs.to_usize());
+            let rhs = get(rhs.to_usize());
+            cmp(&lhs, &rhs)
+        };
+        before.sort_unstable_by(compare);
+    }
+}
+
+/// # Safety
+/// This function guarantees that:
+/// * `get` is only called for `0 <= i < limit`
+/// * `cmp` is only called from the co-domain of `get`.
+#[inline]
+fn sort_unstable_by<I, T, G, F>(
+    indices: &mut [I],
+    get: G,
+    mut cmp: F,
+    descending: bool,
+    limit: usize,
+) where
+    I: Index,
+    G: Fn(usize) -> T,
+    F: FnMut(&T, &T) -> std::cmp::Ordering,
+{
+    if limit != indices.len() {
+        return k_element_sort_inner(indices, get, descending, limit, cmp);
+    }
+
+    if descending {
+        indices.sort_unstable_by(|lhs, rhs| {
+            let lhs = get(lhs.to_usize());
+            let rhs = get(rhs.to_usize());
+            cmp(&lhs, &rhs).reverse()
+        })
+    } else {
+        indices.sort_unstable_by(|lhs, rhs| {
+            let lhs = get(lhs.to_usize());
+            let rhs = get(rhs.to_usize());
+            cmp(&lhs, &rhs)
+        })
+    }
+}
+
+/// # Safety
+/// This function guarantees that:
+/// * `get` is only called for `0 <= i < length`
+/// * `cmp` is only called from the co-domain of `get`.
+#[inline]
+pub(super) fn indices_sorted_unstable_by<I, T, G, F>(
+    validity: &Option<Bitmap>,
+    get: G,
+    cmp: F,
+    length: usize,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> PrimitiveArray<I>
+where
+    I: Index,
+    G: Fn(usize) -> T,
+    F: Fn(&T, &T) -> std::cmp::Ordering,
+{
+    let descending = options.descending;
+
+    let limit = limit.unwrap_or(length);
+    // Safety: without this, we go out of bounds when limit >= length.
+    let limit = limit.min(length);
+
+    let indices = if let Some(validity) = validity {
+        let mut indices = MutableBuffer::<I>::from_len_zeroed(length);
+
+        if options.nulls_first {
+            let mut nulls = 0;
+            let mut valids = 0;
+            validity
+                .iter()
+                .zip(0..length)
+                .for_each(|(is_valid, index)| {
+                    if is_valid {
+                        indices[validity.null_count() + valids] = I::from_usize(index).unwrap();
+                        valids += 1;
+                    } else {
+                        indices[nulls] = I::from_usize(index).unwrap();
+                        nulls += 1;
+                    }
+                });
+
+            if limit > validity.null_count() {
+                // when limit is larger, we must sort values:
+
+                // Soundness:
+                // all indices in `indices` are by construction `< array.len() == values.len()`
+                // limit is by construction < indices.len()
+                let limit = limit - validity.null_count();
+                let indices = &mut indices.as_mut_slice()[validity.null_count()..];
+                sort_unstable_by(indices, get, cmp, options.descending, limit)
+            }
+        } else {
+            let last_valid_index = length - validity.null_count();
+            let mut nulls = 0;
+            let mut valids = 0;
+            validity.iter().zip(0..length).for_each(|(x, index)| {
+                if x {
+                    indices[valids] = I::from_usize(index).unwrap();
+                    valids += 1;
+                } else {
+                    indices[last_valid_index + nulls] = I::from_usize(index).unwrap();
+                    nulls += 1;
+                }
+            });
+
+            // Soundness:
+            // all indices in `indices` are by construction `< array.len() == values.len()`
+            // limit is by construction <= values.len()
+            let limit = limit.min(last_valid_index);
+            let indices = &mut indices.as_mut_slice()[..last_valid_index];
+            sort_unstable_by(indices, get, cmp, options.descending, limit);
+        }
+
+        indices.truncate(limit);
+        indices.shrink_to_fit();
+
+        indices
+    } else {
+        let mut indices = unsafe {
+            MutableBuffer::from_trusted_len_iter_unchecked(
+                (0..length).map(|x| I::from_usize(x).unwrap()),
+            )
+        };
+
+        // Soundness:
+        // indices are by construction `< values.len()`
+        // limit is by construction `< values.len()`
+        sort_unstable_by(&mut indices, get, cmp, descending, limit);
+
+        indices.truncate(limit);
+        indices.shrink_to_fit();
+
+        indices
+    };
+    PrimitiveArray::<I>::from_data(I::DATA_TYPE, indices.into(), None)
+}
diff --git a/src/compute/sort/lex_sort.rs b/src/compute/sort/lex_sort.rs
index 44f0616931..0a596bfd06 100644
--- a/src/compute/sort/lex_sort.rs
+++ b/src/compute/sort/lex_sort.rs
@@ -1,28 +1,10 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements.  See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership.  The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License.  You may obtain a copy of the License at
-//
-//   http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
 use std::cmp::Ordering;
 
 use crate::compute::take;
 use crate::error::{ArrowError, Result};
 use crate::{
-    array::{ord, Array, PrimitiveArray},
+    array::{ord, Array, Index, PrimitiveArray},
     buffer::MutableBuffer,
-    datatypes::DataType,
 };
 
 use super::{sort_to_indices, SortOptions};
@@ -54,7 +36,7 @@ pub struct SortColumn<'a> {
 /// let int64 = Int64Array::from(&[None, Some(-2), Some(89), Some(-64), Some(101)]);
 /// let utf8 = Utf8Array::<i32>::from(&vec![Some("hello"), Some("world"), Some(","), Some("foobar"), Some("!")]);
 ///
-/// let sorted_columns = lexsort(&vec![
+/// let sorted_columns = lexsort::<i32>(&vec![
 ///     SortColumn {
 ///         values: &int64,
 ///         options: None,
@@ -66,14 +48,17 @@ pub struct SortColumn<'a> {
 ///             nulls_first: false,
 ///         }),
 ///     },
-/// ]).unwrap();
+/// ], None).unwrap();
 ///
 /// let sorted = sorted_columns[0].as_any().downcast_ref::<Int64Array>().unwrap();
 /// assert_eq!(sorted.value(1), -64);
 /// assert!(sorted.is_null(0));
 /// ```
-pub fn lexsort(columns: &[SortColumn]) -> Result<Vec<Box<dyn Array>>> {
-    let indices = lexsort_to_indices(columns)?;
+pub fn lexsort<I: Index>(
+    columns: &[SortColumn],
+    limit: Option<usize>,
+) -> Result<Vec<Box<dyn Array>>> {
+    let indices = lexsort_to_indices::<I>(columns, limit)?;
     columns
         .iter()
         .map(|c| take::take(c.values, &indices))
@@ -135,9 +120,12 @@ pub(crate) fn build_compare(array: &dyn Array, sort_option: SortOptions) -> Resu
     })
 }
 
-/// Sort elements lexicographically from a list of `ArrayRef` into an unsigned integer
-/// [`Int32Array`] of indices.
-pub fn lexsort_to_indices(columns: &[SortColumn]) -> Result<PrimitiveArray<i32>> {
+/// Sorts a list of [`SortColumn`] into a non-nullable [`PrimitiveArray`]
+/// representing the indices that would sort the columns.
+pub fn lexsort_to_indices<I: Index>(
+    columns: &[SortColumn],
+    limit: Option<usize>,
+) -> Result<PrimitiveArray<I>> {
     if columns.is_empty() {
         return Err(ArrowError::InvalidArgumentError(
             "Sort requires at least one column".to_string(),
@@ -146,7 +134,7 @@ pub fn lexsort_to_indices(columns: &[SortColumn]) -> Result<PrimitiveArray<i32>>
     if columns.len() == 1 {
         // fallback to non-lexical sort
         let column = &columns[0];
-        return sort_to_indices(column.values, &column.options.unwrap_or_default());
+        return sort_to_indices(column.values, &column.options.unwrap_or_default(), limit);
     }
 
     let row_count = columns[0].values.len();
@@ -164,9 +152,9 @@ pub fn lexsort_to_indices(columns: &[SortColumn]) -> Result<PrimitiveArray<i32>>
         })
         .collect::<Result<Vec<Compare>>>()?;
 
-    let lex_comparator = |a_idx: &i32, b_idx: &i32| -> Ordering {
-        let a_idx = *a_idx as usize;
-        let b_idx = *b_idx as usize;
+    let lex_comparator = |a_idx: &I, b_idx: &I| -> Ordering {
+        let a_idx = a_idx.to_usize();
+        let b_idx = b_idx.to_usize();
         for comparator in comparators.iter() {
             match comparator(a_idx, b_idx) {
                 Ordering::Equal => continue,
@@ -178,12 +166,23 @@ pub fn lexsort_to_indices(columns: &[SortColumn]) -> Result<PrimitiveArray<i32>>
     };
 
     // Safety: `0..row_count` is TrustedLen
-    let mut values =
-        unsafe { MutableBuffer::<i32>::from_trusted_len_iter_unchecked(0..row_count as i32) };
-    values.sort_unstable_by(lex_comparator);
+    let mut values = unsafe {
+        MutableBuffer::from_trusted_len_iter_unchecked(
+            (0..row_count).map(|x| I::from_usize(x).unwrap()),
+        )
+    };
 
-    Ok(PrimitiveArray::<i32>::from_data(
-        DataType::Int32,
+    if let Some(limit) = limit {
+        let limit = limit.min(row_count);
+        let (before, _, _) = values.select_nth_unstable_by(limit, lex_comparator);
+        before.sort_unstable_by(lex_comparator);
+        values.truncate(limit);
+    } else {
+        values.sort_unstable_by(lex_comparator);
+    }
+
+    Ok(PrimitiveArray::<I>::from_data(
+        I::DATA_TYPE,
         values.into(),
         None,
     ))
@@ -196,7 +195,14 @@ mod tests {
     use super::*;
 
     fn test_lex_sort_arrays(input: Vec<SortColumn>, expected: Vec<Box<dyn Array>>) {
-        let sorted = lexsort(&input).unwrap();
+        let sorted = lexsort::<i32>(&input, None).unwrap();
+        assert_eq!(sorted, expected);
+
+        let sorted = lexsort::<i32>(&input, Some(2)).unwrap();
+        let expected = expected
+            .into_iter()
+            .map(|x| x.slice(0, 2))
+            .collect::<Vec<_>>();
         assert_eq!(sorted, expected);
     }
 
diff --git a/src/compute/sort/mod.rs b/src/compute/sort/mod.rs
index 1273450569..32cf63166c 100644
--- a/src/compute/sort/mod.rs
+++ b/src/compute/sort/mod.rs
@@ -1,23 +1,4 @@
-// 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.
-
-//! Defines sort kernel for `ArrayRef`
-
-use std::cmp::{Ordering, Reverse};
+use std::cmp::Ordering;
 
 use crate::array::ord;
 use crate::compute::take;
@@ -25,136 +6,158 @@ use crate::datatypes::*;
 use crate::error::{ArrowError, Result};
 use crate::{
     array::*,
-    buffer::Buffer,
     types::{days_ms, NativeType},
 };
 
 use crate::buffer::MutableBuffer;
-use num::ToPrimitive;
 
+mod boolean;
+mod common;
 mod lex_sort;
 mod primitive;
+mod utf8;
 
 pub(crate) use lex_sort::{build_compare, Compare};
 pub use lex_sort::{lexsort, lexsort_to_indices, SortColumn};
 
 macro_rules! dyn_sort {
-    ($ty:ty, $array:expr, $cmp:expr, $options:expr) => {{
+    ($ty:ty, $array:expr, $cmp:expr, $options:expr, $limit:expr) => {{
         let array = $array
             .as_any()
             .downcast_ref::<PrimitiveArray<$ty>>()
             .unwrap();
         Ok(Box::new(primitive::sort_by::<$ty, _>(
-            &array, $cmp, $options,
+            &array, $cmp, $options, $limit,
         )))
     }};
 }
 
-/// Sort the `ArrayRef` using `SortOptions`.
+/// Sort the [`Array`] using [`SortOptions`].
 ///
-/// Performs a stable sort on values and indices. Nulls are ordered according to the `nulls_first` flag in `options`.
+/// Performs an unstable sort on values and indices. Nulls are ordered according to the `nulls_first` flag in `options`.
 /// Floats are sorted using IEEE 754 totalOrder
-///
-/// Returns an `ArrowError::ComputeError(String)` if the array type is either unsupported by `sort_to_indices` or `take`.
-///
-pub fn sort(values: &dyn Array, options: &SortOptions) -> Result<Box<dyn Array>> {
+/// # Errors
+/// Errors if the [`DataType`] is not supported.
+pub fn sort(
+    values: &dyn Array,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> Result<Box<dyn Array>> {
     match values.data_type() {
-        DataType::Int8 => dyn_sort!(i8, values, ord::total_cmp, options),
-        DataType::Int16 => dyn_sort!(i16, values, ord::total_cmp, options),
+        DataType::Int8 => dyn_sort!(i8, values, ord::total_cmp, options, limit),
+        DataType::Int16 => dyn_sort!(i16, values, ord::total_cmp, options, limit),
         DataType::Int32
         | DataType::Date32
         | DataType::Time32(_)
         | DataType::Interval(IntervalUnit::YearMonth) => {
-            dyn_sort!(i32, values, ord::total_cmp, options)
+            dyn_sort!(i32, values, ord::total_cmp, options, limit)
         }
         DataType::Int64
         | DataType::Date64
         | DataType::Time64(_)
         | DataType::Timestamp(_, None)
-        | DataType::Duration(_) => dyn_sort!(i64, values, ord::total_cmp, options),
-        DataType::UInt8 => dyn_sort!(u8, values, ord::total_cmp, options),
-        DataType::UInt16 => dyn_sort!(u16, values, ord::total_cmp, options),
-        DataType::UInt32 => dyn_sort!(u32, values, ord::total_cmp, options),
-        DataType::UInt64 => dyn_sort!(u64, values, ord::total_cmp, options),
-        DataType::Float32 => dyn_sort!(f32, values, ord::total_cmp_f32, options),
-        DataType::Float64 => dyn_sort!(f64, values, ord::total_cmp_f64, options),
+        | DataType::Duration(_) => dyn_sort!(i64, values, ord::total_cmp, options, limit),
+        DataType::UInt8 => dyn_sort!(u8, values, ord::total_cmp, options, limit),
+        DataType::UInt16 => dyn_sort!(u16, values, ord::total_cmp, options, limit),
+        DataType::UInt32 => dyn_sort!(u32, values, ord::total_cmp, options, limit),
+        DataType::UInt64 => dyn_sort!(u64, values, ord::total_cmp, options, limit),
+        DataType::Float32 => dyn_sort!(f32, values, ord::total_cmp_f32, options, limit),
+        DataType::Float64 => dyn_sort!(f64, values, ord::total_cmp_f64, options, limit),
         DataType::Interval(IntervalUnit::DayTime) => {
-            dyn_sort!(days_ms, values, ord::total_cmp, options)
+            dyn_sort!(days_ms, values, ord::total_cmp, options, limit)
         }
         _ => {
-            let indices = sort_to_indices(values, options)?;
+            let indices = sort_to_indices::<u64>(values, options, limit)?;
             take::take(values, &indices)
         }
     }
 }
 
 // partition indices into valid and null indices
-fn partition_validity(array: &dyn Array) -> (Vec<i32>, Vec<i32>) {
-    let indices = 0..(array.len().to_i32().unwrap());
-    indices.partition(|index| array.is_valid(*index as usize))
+fn partition_validity<I: Index>(array: &dyn Array) -> (Vec<I>, Vec<I>) {
+    let length = array.len();
+    let indices = (0..length).map(|x| I::from_usize(x).unwrap());
+    if let Some(validity) = array.validity() {
+        indices.partition(|index| validity.get_bit(index.to_usize()))
+    } else {
+        (indices.collect(), vec![])
+    }
 }
 
 macro_rules! dyn_sort_indices {
-    ($ty:ty, $array:expr, $cmp:expr, $options:expr) => {{
+    ($index:ty, $ty:ty, $array:expr, $cmp:expr, $options:expr, $limit:expr) => {{
         let array = $array
             .as_any()
             .downcast_ref::<PrimitiveArray<$ty>>()
             .unwrap();
-        Ok(primitive::indices_sorted_by::<$ty, _>(
-            &array, $cmp, $options,
+        Ok(primitive::indices_sorted_unstable_by::<$index, $ty, _>(
+            &array, $cmp, $options, $limit,
         ))
     }};
 }
 
-/// Sort elements from `ArrayRef` into an unsigned integer (`UInt32Array`) of indices.
-/// For floating point arrays any NaN values are considered to be greater than any other non-null value
-pub fn sort_to_indices(values: &dyn Array, options: &SortOptions) -> Result<Int32Array> {
+/// Sort elements from `values` into a non-nullable [`PrimitiveArray`] of indices that sort `values`.
+pub fn sort_to_indices<I: Index>(
+    values: &dyn Array,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> Result<PrimitiveArray<I>> {
     match values.data_type() {
         DataType::Boolean => {
             let (v, n) = partition_validity(values);
-            Ok(sort_boolean(values, v, n, &options))
+            Ok(boolean::sort_boolean(
+                values.as_any().downcast_ref().unwrap(),
+                v,
+                n,
+                options,
+                limit,
+            ))
         }
-        DataType::Int8 => dyn_sort_indices!(i8, values, ord::total_cmp, options),
-        DataType::Int16 => dyn_sort_indices!(i16, values, ord::total_cmp, options),
+        DataType::Int8 => dyn_sort_indices!(I, i8, values, ord::total_cmp, options, limit),
+        DataType::Int16 => dyn_sort_indices!(I, i16, values, ord::total_cmp, options, limit),
         DataType::Int32
         | DataType::Date32
         | DataType::Time32(_)
         | DataType::Interval(IntervalUnit::YearMonth) => {
-            dyn_sort_indices!(i32, values, ord::total_cmp, options)
+            dyn_sort_indices!(I, i32, values, ord::total_cmp, options, limit)
         }
         DataType::Int64
         | DataType::Date64
         | DataType::Time64(_)
         | DataType::Timestamp(_, None)
-        | DataType::Duration(_) => dyn_sort_indices!(i64, values, ord::total_cmp, options),
-        DataType::UInt8 => dyn_sort_indices!(u8, values, ord::total_cmp, options),
-        DataType::UInt16 => dyn_sort_indices!(u16, values, ord::total_cmp, options),
-        DataType::UInt32 => dyn_sort_indices!(u32, values, ord::total_cmp, options),
-        DataType::UInt64 => dyn_sort_indices!(u64, values, ord::total_cmp, options),
-        DataType::Float32 => dyn_sort_indices!(f32, values, ord::total_cmp_f32, options),
-        DataType::Float64 => dyn_sort_indices!(f64, values, ord::total_cmp_f64, options),
-        DataType::Interval(IntervalUnit::DayTime) => {
-            dyn_sort_indices!(days_ms, values, ord::total_cmp, options)
-        }
-        DataType::Utf8 => {
-            let (v, n) = partition_validity(values);
-            Ok(sort_utf8::<i32>(values, v, n, &options))
+        | DataType::Duration(_) => {
+            dyn_sort_indices!(I, i64, values, ord::total_cmp, options, limit)
         }
-        DataType::LargeUtf8 => {
-            let (v, n) = partition_validity(values);
-            Ok(sort_utf8::<i64>(values, v, n, &options))
+        DataType::UInt8 => dyn_sort_indices!(I, u8, values, ord::total_cmp, options, limit),
+        DataType::UInt16 => dyn_sort_indices!(I, u16, values, ord::total_cmp, options, limit),
+        DataType::UInt32 => dyn_sort_indices!(I, u32, values, ord::total_cmp, options, limit),
+        DataType::UInt64 => dyn_sort_indices!(I, u64, values, ord::total_cmp, options, limit),
+        DataType::Float32 => dyn_sort_indices!(I, f32, values, ord::total_cmp_f32, options, limit),
+        DataType::Float64 => dyn_sort_indices!(I, f64, values, ord::total_cmp_f64, options, limit),
+        DataType::Interval(IntervalUnit::DayTime) => {
+            dyn_sort_indices!(I, days_ms, values, ord::total_cmp, options, limit)
         }
+        DataType::Utf8 => Ok(utf8::indices_sorted_unstable_by::<I, i32>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::LargeUtf8 => Ok(utf8::indices_sorted_unstable_by::<I, i64>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
         DataType::List(field) => {
             let (v, n) = partition_validity(values);
             match field.data_type() {
-                DataType::Int8 => Ok(sort_list::<i32, i8>(values, v, n, &options)),
-                DataType::Int16 => Ok(sort_list::<i32, i16>(values, v, n, &options)),
-                DataType::Int32 => Ok(sort_list::<i32, i32>(values, v, n, &options)),
-                DataType::Int64 => Ok(sort_list::<i32, i64>(values, v, n, &options)),
-                DataType::UInt8 => Ok(sort_list::<i32, u8>(values, v, n, &options)),
-                DataType::UInt16 => Ok(sort_list::<i32, u16>(values, v, n, &options)),
-                DataType::UInt32 => Ok(sort_list::<i32, u32>(values, v, n, &options)),
-                DataType::UInt64 => Ok(sort_list::<i32, u64>(values, v, n, &options)),
+                DataType::Int8 => Ok(sort_list::<I, i32, i8>(values, v, n, options, limit)),
+                DataType::Int16 => Ok(sort_list::<I, i32, i16>(values, v, n, options, limit)),
+                DataType::Int32 => Ok(sort_list::<I, i32, i32>(values, v, n, options, limit)),
+                DataType::Int64 => Ok(sort_list::<I, i32, i64>(values, v, n, options, limit)),
+                DataType::UInt8 => Ok(sort_list::<I, i32, u8>(values, v, n, options, limit)),
+                DataType::UInt16 => Ok(sort_list::<I, i32, u16>(values, v, n, options, limit)),
+                DataType::UInt32 => Ok(sort_list::<I, i32, u32>(values, v, n, options, limit)),
+                DataType::UInt64 => Ok(sort_list::<I, i32, u64>(values, v, n, options, limit)),
                 t => Err(ArrowError::NotYetImplemented(format!(
                     "Sort not supported for list type {:?}",
                     t
@@ -164,14 +167,14 @@ pub fn sort_to_indices(values: &dyn Array, options: &SortOptions) -> Result<Int3
         DataType::LargeList(field) => {
             let (v, n) = partition_validity(values);
             match field.data_type() {
-                DataType::Int8 => Ok(sort_list::<i64, i8>(values, v, n, &options)),
-                DataType::Int16 => Ok(sort_list::<i64, i16>(values, v, n, &options)),
-                DataType::Int32 => Ok(sort_list::<i64, i32>(values, v, n, &options)),
-                DataType::Int64 => Ok(sort_list::<i64, i64>(values, v, n, &options)),
-                DataType::UInt8 => Ok(sort_list::<i64, u8>(values, v, n, &options)),
-                DataType::UInt16 => Ok(sort_list::<i64, u16>(values, v, n, &options)),
-                DataType::UInt32 => Ok(sort_list::<i64, u32>(values, v, n, &options)),
-                DataType::UInt64 => Ok(sort_list::<i64, u64>(values, v, n, &options)),
+                DataType::Int8 => Ok(sort_list::<I, i64, i8>(values, v, n, options, limit)),
+                DataType::Int16 => Ok(sort_list::<I, i64, i16>(values, v, n, options, limit)),
+                DataType::Int32 => Ok(sort_list::<I, i64, i32>(values, v, n, options, limit)),
+                DataType::Int64 => Ok(sort_list::<I, i64, i64>(values, v, n, options, limit)),
+                DataType::UInt8 => Ok(sort_list::<I, i64, u8>(values, v, n, options, limit)),
+                DataType::UInt16 => Ok(sort_list::<I, i64, u16>(values, v, n, options, limit)),
+                DataType::UInt32 => Ok(sort_list::<I, i64, u32>(values, v, n, options, limit)),
+                DataType::UInt64 => Ok(sort_list::<I, i64, u64>(values, v, n, options, limit)),
                 t => Err(ArrowError::NotYetImplemented(format!(
                     "Sort not supported for list type {:?}",
                     t
@@ -181,37 +184,28 @@ pub fn sort_to_indices(values: &dyn Array, options: &SortOptions) -> Result<Int3
         DataType::FixedSizeList(field, _) => {
             let (v, n) = partition_validity(values);
             match field.data_type() {
-                DataType::Int8 => Ok(sort_list::<i32, i8>(values, v, n, &options)),
-                DataType::Int16 => Ok(sort_list::<i32, i16>(values, v, n, &options)),
-                DataType::Int32 => Ok(sort_list::<i32, i32>(values, v, n, &options)),
-                DataType::Int64 => Ok(sort_list::<i32, i64>(values, v, n, &options)),
-                DataType::UInt8 => Ok(sort_list::<i32, u8>(values, v, n, &options)),
-                DataType::UInt16 => Ok(sort_list::<i32, u16>(values, v, n, &options)),
-                DataType::UInt32 => Ok(sort_list::<i32, u32>(values, v, n, &options)),
-                DataType::UInt64 => Ok(sort_list::<i32, u64>(values, v, n, &options)),
+                DataType::Int8 => Ok(sort_list::<I, i32, i8>(values, v, n, options, limit)),
+                DataType::Int16 => Ok(sort_list::<I, i32, i16>(values, v, n, options, limit)),
+                DataType::Int32 => Ok(sort_list::<I, i32, i32>(values, v, n, options, limit)),
+                DataType::Int64 => Ok(sort_list::<I, i32, i64>(values, v, n, options, limit)),
+                DataType::UInt8 => Ok(sort_list::<I, i32, u8>(values, v, n, options, limit)),
+                DataType::UInt16 => Ok(sort_list::<I, i32, u16>(values, v, n, options, limit)),
+                DataType::UInt32 => Ok(sort_list::<I, i32, u32>(values, v, n, options, limit)),
+                DataType::UInt64 => Ok(sort_list::<I, i32, u64>(values, v, n, options, limit)),
                 t => Err(ArrowError::NotYetImplemented(format!(
                     "Sort not supported for list type {:?}",
                     t
                 ))),
             }
         }
-        DataType::Dictionary(key_type, value_type) if *value_type.as_ref() == DataType::Utf8 => {
-            let (v, n) = partition_validity(values);
-            match key_type.as_ref() {
-                DataType::Int8 => Ok(sort_string_dictionary::<i8>(values, v, n, &options)),
-                DataType::Int16 => Ok(sort_string_dictionary::<i16>(values, v, n, &options)),
-                DataType::Int32 => Ok(sort_string_dictionary::<i32>(values, v, n, &options)),
-                DataType::Int64 => Ok(sort_string_dictionary::<i64>(values, v, n, &options)),
-                DataType::UInt8 => Ok(sort_string_dictionary::<u8>(values, v, n, &options)),
-                DataType::UInt16 => Ok(sort_string_dictionary::<u16>(values, v, n, &options)),
-                DataType::UInt32 => Ok(sort_string_dictionary::<u32>(values, v, n, &options)),
-                DataType::UInt64 => Ok(sort_string_dictionary::<u64>(values, v, n, &options)),
-                t => Err(ArrowError::NotYetImplemented(format!(
-                    "Sort not supported for dictionary key type {:?}",
-                    t
-                ))),
-            }
-        }
+        DataType::Dictionary(key_type, value_type) => match value_type.as_ref() {
+            DataType::Utf8 => sort_dict::<I, i32>(values, key_type.as_ref(), options, limit),
+            DataType::LargeUtf8 => sort_dict::<I, i64>(values, key_type.as_ref(), options, limit),
+            t => Err(ArrowError::NotYetImplemented(format!(
+                "Sort not supported for dictionary type with keys {:?}",
+                t
+            ))),
+        },
         t => Err(ArrowError::NotYetImplemented(format!(
             "Sort not supported for data type {:?}",
             t
@@ -219,6 +213,60 @@ pub fn sort_to_indices(values: &dyn Array, options: &SortOptions) -> Result<Int3
     }
 }
 
+fn sort_dict<I: Index, O: Offset>(
+    values: &dyn Array,
+    key_type: &DataType,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> Result<PrimitiveArray<I>> {
+    match key_type {
+        DataType::Int8 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, i8, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::Int16 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, i16, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::Int32 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, i32, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::Int64 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, i64, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::UInt8 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, u8, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::UInt16 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, u16, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::UInt32 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, u32, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        DataType::UInt64 => Ok(utf8::indices_sorted_unstable_by_dictionary::<I, u64, O>(
+            values.as_any().downcast_ref().unwrap(),
+            options,
+            limit,
+        )),
+        t => Err(ArrowError::NotYetImplemented(format!(
+            "Sort not supported for dictionary key type {:?}",
+            t
+        ))),
+    }
+}
+
 /// Checks if an array of type `datatype` can be sorted
 ///
 /// # Examples
@@ -303,144 +351,19 @@ impl Default for SortOptions {
     }
 }
 
-/// Sort primitive values
-fn sort_boolean(
-    values: &dyn Array,
-    value_indices: Vec<i32>,
-    null_indices: Vec<i32>,
-    options: &SortOptions,
-) -> Int32Array {
-    let values = values
-        .as_any()
-        .downcast_ref::<BooleanArray>()
-        .expect("Unable to downcast to boolean array");
-    let descending = options.descending;
-
-    // create tuples that are used for sorting
-    let mut valids = value_indices
-        .into_iter()
-        .map(|index| (index, values.value(index as usize)))
-        .collect::<Vec<(i32, bool)>>();
-
-    let mut nulls = null_indices;
-
-    if !descending {
-        valids.sort_by(|a, b| a.1.cmp(&b.1));
-    } else {
-        valids.sort_by(|a, b| a.1.cmp(&b.1).reverse());
-        // reverse to keep a stable ordering
-        nulls.reverse();
-    }
-
-    let mut values = MutableBuffer::<i32>::with_capacity(values.len());
-
-    if options.nulls_first {
-        values.extend_from_slice(nulls.as_slice());
-        valids.iter().for_each(|x| values.push(x.0));
-    } else {
-        // nulls last
-        valids.iter().for_each(|x| values.push(x.0));
-        values.extend_from_slice(nulls.as_slice());
-    }
-
-    Int32Array::from_data(DataType::Int32, values.into(), None)
-}
-
-/// Sort strings
-fn sort_utf8<O: Offset>(
-    values: &dyn Array,
-    value_indices: Vec<i32>,
-    null_indices: Vec<i32>,
-    options: &SortOptions,
-) -> Int32Array {
-    let values = values.as_any().downcast_ref::<Utf8Array<O>>().unwrap();
-
-    sort_string_helper(
-        values,
-        value_indices,
-        null_indices,
-        options,
-        |array, idx| array.value(idx as usize),
-    )
-}
-
-/// Sort dictionary encoded strings
-fn sort_string_dictionary<T: DictionaryKey>(
-    values: &dyn Array,
-    value_indices: Vec<i32>,
-    null_indices: Vec<i32>,
-    options: &SortOptions,
-) -> Int32Array {
-    let values: &DictionaryArray<T> = values
-        .as_any()
-        .downcast_ref::<DictionaryArray<T>>()
-        .unwrap();
-
-    let keys = values.keys();
-
-    let dict = values.values();
-    let dict = dict.as_any().downcast_ref::<Utf8Array<i32>>().unwrap();
-
-    sort_string_helper(
-        keys,
-        value_indices,
-        null_indices,
-        options,
-        |array: &PrimitiveArray<T>, idx| -> &str {
-            let key: T = array.value(idx as usize);
-            dict.value(key.to_usize().unwrap())
-        },
-    )
-}
-
-/// shared implementation between dictionary encoded and plain string arrays
-#[inline]
-fn sort_string_helper<'a, A: Array, F>(
-    values: &'a A,
-    value_indices: Vec<i32>,
-    null_indices: Vec<i32>,
-    options: &SortOptions,
-    value_fn: F,
-) -> Int32Array
-where
-    F: Fn(&'a A, i32) -> &str,
-{
-    let mut valids = value_indices
-        .into_iter()
-        .map(|index| (index, value_fn(&values, index)))
-        .collect::<Vec<(i32, &str)>>();
-    let mut nulls = null_indices;
-    if !options.descending {
-        valids.sort_by_key(|a| a.1);
-    } else {
-        valids.sort_by_key(|a| Reverse(a.1));
-        nulls.reverse();
-    }
-
-    let valids = valids.iter().map(|tuple| tuple.0);
-
-    let values = if options.nulls_first {
-        let values = nulls.into_iter().chain(valids);
-        Buffer::<i32>::from_trusted_len_iter(values)
-    } else {
-        let values = valids.chain(nulls.into_iter());
-        Buffer::<i32>::from_trusted_len_iter(values)
-    };
-
-    PrimitiveArray::<i32>::from_data(DataType::Int32, values, None)
-}
-
-fn sort_list<O, T>(
+fn sort_list<I, O, T>(
     values: &dyn Array,
-    value_indices: Vec<i32>,
-    null_indices: Vec<i32>,
+    value_indices: Vec<I>,
+    null_indices: Vec<I>,
     options: &SortOptions,
-) -> Int32Array
+    limit: Option<usize>,
+) -> PrimitiveArray<I>
 where
+    I: Index,
     O: Offset,
     T: NativeType + std::cmp::PartialOrd,
 {
-    let mut valids: Vec<(i32, Box<dyn Array>)> = values
+    let mut valids: Vec<(I, Box<dyn Array>)> = values
         .as_any()
         .downcast_ref::<FixedSizeListArray>()
         .map_or_else(
@@ -449,14 +372,14 @@ where
                 value_indices
                     .iter()
                     .copied()
-                    .map(|index| (index, values.value(index as usize)))
+                    .map(|index| (index, values.value(index.to_usize())))
                     .collect()
             },
             |values| {
                 value_indices
                     .iter()
                     .copied()
-                    .map(|index| (index, values.value(index as usize)))
+                    .map(|index| (index, values.value(index.to_usize())))
                     .collect()
             },
         );
@@ -469,17 +392,19 @@ where
 
     let values = valids.iter().map(|tuple| tuple.0);
 
-    let values = if options.nulls_first {
-        let mut buffer = MutableBuffer::<i32>::from_trusted_len_iter(null_indices.into_iter());
-        values.for_each(|x| buffer.push(x));
-        buffer.into()
+    let mut values = if options.nulls_first {
+        let mut buffer = MutableBuffer::<I>::from_trusted_len_iter(null_indices.into_iter());
+        buffer.extend(values);
+        buffer
     } else {
-        let mut buffer = MutableBuffer::<i32>::from_trusted_len_iter(values);
-        null_indices.iter().for_each(|x| buffer.push(*x));
-        buffer.into()
+        let mut buffer = MutableBuffer::<I>::from_trusted_len_iter(values);
+        buffer.extend(null_indices);
+        buffer
     };
 
-    PrimitiveArray::<i32>::from_data(DataType::Int32, values, None)
+    values.truncate(limit.unwrap_or_else(|| values.len()));
+
+    PrimitiveArray::<I>::from_data(I::DATA_TYPE, values.into(), None)
 }
 
 /// Compare two `Array`s based on the ordering defined in [ord](crate::array::ord).
@@ -507,7 +432,7 @@ mod tests {
     ) {
         let output = BooleanArray::from(data);
         let expected = Int32Array::from_slice(expected_data);
-        let output = sort_to_indices(&output, &options).unwrap();
+        let output = sort_to_indices(&output, &options, None).unwrap();
         assert_eq!(output, expected)
     }
 
@@ -521,7 +446,7 @@ mod tests {
     {
         let input = PrimitiveArray::<T>::from(data).to(data_type.clone());
         let expected = PrimitiveArray::<T>::from(expected_data).to(data_type);
-        let output = sort(&input, &options).unwrap();
+        let output = sort(&input, &options, None).unwrap();
         assert_eq!(expected, output.as_ref())
     }
 
@@ -532,7 +457,7 @@ mod tests {
     ) {
         let input = Utf8Array::<i32>::from(&data.to_vec());
         let expected = Int32Array::from_slice(expected_data);
-        let output = sort_to_indices(&input, &options).unwrap();
+        let output = sort_to_indices(&input, &options, None).unwrap();
         assert_eq!(output, expected)
     }
 
@@ -543,7 +468,7 @@ mod tests {
     ) {
         let input = Utf8Array::<i32>::from(&data.to_vec());
         let expected = Utf8Array::<i32>::from(&expected_data.to_vec());
-        let output = sort(&input, &options).unwrap();
+        let output = sort(&input, &options, None).unwrap();
         assert_eq!(expected, output.as_ref())
     }
 
@@ -560,7 +485,7 @@ mod tests {
         expected.try_extend(expected_data.iter().copied()).unwrap();
         let expected = expected.into_arc();
 
-        let output = sort(input.as_ref(), &options).unwrap();
+        let output = sort(input.as_ref(), &options, None).unwrap();
         assert_eq!(expected.as_ref(), output.as_ref())
     }
 
@@ -704,6 +629,7 @@ mod tests {
                 descending: false,
                 nulls_first: true,
             },
+            // &[3, 0, 5, 1, 4, 2] is also valid
             &[0, 3, 5, 1, 4, 2],
         );
 
@@ -720,7 +646,8 @@ mod tests {
                 descending: true,
                 nulls_first: false,
             },
-            &[2, 4, 1, 5, 3, 0],
+            // &[2, 4, 1, 5, 3, 0] is also valid
+            &[2, 4, 1, 5, 0, 3],
         );
 
         test_sort_to_indices_string_arrays(
@@ -736,6 +663,7 @@ mod tests {
                 descending: false,
                 nulls_first: true,
             },
+            // &[3, 0, 5, 1, 4, 2] is also valid
             &[0, 3, 5, 1, 4, 2],
         );
 
@@ -752,7 +680,8 @@ mod tests {
                 descending: true,
                 nulls_first: true,
             },
-            &[3, 0, 2, 4, 1, 5],
+            // &[3, 0, 2, 4, 1, 5] is also valid
+            &[0, 3, 2, 4, 1, 5],
         );
     }
 
@@ -1095,18 +1024,14 @@ mod tests {
 
         datatypes.into_iter().for_each(|d1| {
             let array = new_null_array(d1.clone(), 10);
+            let options = SortOptions {
+                descending: true,
+                nulls_first: true,
+            };
             if can_sort(&d1) {
-                let options = SortOptions {
-                    descending: true,
-                    nulls_first: true,
-                };
-                assert!(sort(array.as_ref(), &options).is_ok());
+                assert!(sort(array.as_ref(), &options, None).is_ok());
             } else {
-                let options = SortOptions {
-                    descending: true,
-                    nulls_first: true,
-                };
-                assert!(sort(array.as_ref(), &options).is_err());
+                assert!(sort(array.as_ref(), &options, None).is_err());
             }
         });
     }
diff --git a/src/compute/sort/primitive/indices.rs b/src/compute/sort/primitive/indices.rs
index 41dea70c57..e7c79947b4 100644
--- a/src/compute/sort/primitive/indices.rs
+++ b/src/compute/sort/primitive/indices.rs
@@ -1,132 +1,32 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements.  See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership.  The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License.  You may obtain a copy of the License at
-//
-//   http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
 use crate::{
-    array::{Array, PrimitiveArray},
-    buffer::MutableBuffer,
-    datatypes::DataType,
+    array::{Array, Index, PrimitiveArray},
     types::NativeType,
 };
 
+use super::super::common;
 use super::super::SortOptions;
 
-/// # Safety
-/// `indices[i] < values.len()` for all i
-#[inline]
-unsafe fn sort_inner<T, F>(indices: &mut [i32], values: &[T], mut cmp: F, descending: bool)
-where
-    T: NativeType,
-    F: FnMut(&T, &T) -> std::cmp::Ordering,
-{
-    if descending {
-        indices.sort_by(|lhs, rhs| {
-            let lhs = values.get_unchecked(*lhs as usize);
-            let rhs = values.get_unchecked(*rhs as usize);
-            cmp(lhs, rhs).reverse()
-        })
-    } else {
-        indices.sort_by(|lhs, rhs| {
-            let lhs = values.get_unchecked(*lhs as usize);
-            let rhs = values.get_unchecked(*rhs as usize);
-            cmp(lhs, rhs)
-        })
-    }
-}
-
-pub fn indices_sorted_by<T, F>(
+/// Unstable sort of indices.
+pub fn indices_sorted_unstable_by<I, T, F>(
     array: &PrimitiveArray<T>,
     cmp: F,
     options: &SortOptions,
-) -> PrimitiveArray<i32>
+    limit: Option<usize>,
+) -> PrimitiveArray<I>
 where
+    I: Index,
     T: NativeType,
     F: Fn(&T, &T) -> std::cmp::Ordering,
 {
-    let descending = options.descending;
-    let values = array.values();
-    let validity = array.validity();
-
-    if let Some(validity) = validity {
-        let mut indices = MutableBuffer::<i32>::from_len_zeroed(array.len());
-
-        if options.nulls_first {
-            let mut nulls = 0;
-            let mut valids = 0;
-            validity
-                .iter()
-                .zip(0..array.len() as i32)
-                .for_each(|(x, index)| {
-                    if x {
-                        indices[validity.null_count() + valids] = index;
-                        valids += 1;
-                    } else {
-                        indices[nulls] = index;
-                        nulls += 1;
-                    }
-                });
-            // Soundness:
-            // all indices in `indices` are by construction `< array.len() == values.len()`
-            unsafe {
-                sort_inner(
-                    &mut indices.as_mut_slice()[validity.null_count()..],
-                    values,
-                    cmp,
-                    options.descending,
-                )
-            }
-        } else {
-            let last_valid_index = array.len() - validity.null_count();
-            let mut nulls = 0;
-            let mut valids = 0;
-            validity
-                .iter()
-                .zip(0..array.len() as i32)
-                .for_each(|(x, index)| {
-                    if x {
-                        indices[valids] = index;
-                        valids += 1;
-                    } else {
-                        indices[last_valid_index + nulls] = index;
-                        nulls += 1;
-                    }
-                });
-
-            // Soundness:
-            // all indices in `indices` are by construction `< array.len() == values.len()`
-            unsafe {
-                sort_inner(
-                    &mut indices.as_mut_slice()[..last_valid_index],
-                    values,
-                    cmp,
-                    options.descending,
-                )
-            };
-        }
-
-        PrimitiveArray::<i32>::from_data(DataType::Int32, indices.into(), None)
-    } else {
-        let mut indices =
-            unsafe { MutableBuffer::from_trusted_len_iter_unchecked(0..values.len() as i32) };
-
-        // Soundness:
-        // indices are by construction `< values.len()`
-        unsafe { sort_inner(&mut indices, values, cmp, descending) };
-
-        PrimitiveArray::<i32>::from_data(DataType::Int32, indices.into(), None)
+    unsafe {
+        common::indices_sorted_unstable_by(
+            array.validity(),
+            |x: usize| *array.values().as_slice().get_unchecked(x),
+            cmp,
+            array.len(),
+            options,
+            limit,
+        )
     }
 }
 
@@ -135,14 +35,21 @@ mod tests {
     use super::*;
     use crate::array::ord;
     use crate::array::*;
-
-    fn test<T>(data: &[Option<T>], data_type: DataType, options: SortOptions, expected_data: &[i32])
-    where
+    use crate::datatypes::DataType;
+
+    fn test<T>(
+        data: &[Option<T>],
+        data_type: DataType,
+        options: SortOptions,
+        limit: Option<usize>,
+        expected_data: &[i32],
+    ) where
         T: NativeType + std::cmp::Ord,
     {
         let input = PrimitiveArray::<T>::from(data).to(data_type);
         let expected = Int32Array::from_slice(&expected_data);
-        let output = indices_sorted_by(&input, ord::total_cmp, &options);
+        let output =
+            indices_sorted_unstable_by::<i32, _, _>(&input, ord::total_cmp, &options, limit);
         assert_eq!(output, expected)
     }
 
@@ -155,6 +62,7 @@ mod tests {
                 descending: false,
                 nulls_first: true,
             },
+            None,
             &[0, 5, 3, 1, 4, 2],
         );
     }
@@ -168,6 +76,7 @@ mod tests {
                 descending: false,
                 nulls_first: false,
             },
+            None,
             &[3, 1, 4, 2, 0, 5],
         );
     }
@@ -181,6 +90,7 @@ mod tests {
                 descending: true,
                 nulls_first: true,
             },
+            None,
             &[0, 5, 2, 1, 4, 3],
         );
     }
@@ -194,7 +104,116 @@ mod tests {
                 descending: true,
                 nulls_first: false,
             },
+            None,
             &[2, 1, 4, 3, 0, 5],
         );
     }
+
+    #[test]
+    fn limit_ascending_nulls_first() {
+        // nulls sorted
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: false,
+                nulls_first: true,
+            },
+            Some(2),
+            &[0, 5],
+        );
+
+        // nulls and values sorted
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: false,
+                nulls_first: true,
+            },
+            Some(4),
+            &[0, 5, 3, 1],
+        );
+    }
+
+    #[test]
+    fn limit_ascending_nulls_last() {
+        // values
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: false,
+                nulls_first: false,
+            },
+            Some(2),
+            &[3, 1],
+        );
+
+        // values and nulls
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: false,
+                nulls_first: false,
+            },
+            Some(5),
+            &[3, 1, 4, 2, 0],
+        );
+    }
+
+    #[test]
+    fn limit_descending_nulls_first() {
+        // nulls
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: true,
+                nulls_first: true,
+            },
+            Some(2),
+            &[0, 5],
+        );
+
+        // nulls and values
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: true,
+                nulls_first: true,
+            },
+            Some(4),
+            &[0, 5, 2, 1],
+        );
+    }
+
+    #[test]
+    fn limit_descending_nulls_last() {
+        // values
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: true,
+                nulls_first: false,
+            },
+            Some(2),
+            &[2, 1],
+        );
+
+        // values and nulls
+        test::<i8>(
+            &[None, Some(3), Some(5), Some(2), Some(3), None],
+            DataType::Int8,
+            SortOptions {
+                descending: true,
+                nulls_first: false,
+            },
+            Some(5),
+            &[2, 1, 4, 3, 0],
+        );
+    }
 }
diff --git a/src/compute/sort/primitive/mod.rs b/src/compute/sort/primitive/mod.rs
index 6be3f81c96..ccfeb15b86 100644
--- a/src/compute/sort/primitive/mod.rs
+++ b/src/compute/sort/primitive/mod.rs
@@ -1,22 +1,5 @@
-// 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.
-
 mod indices;
 mod sort;
 
-pub use indices::indices_sorted_by;
+pub use indices::indices_sorted_unstable_by;
 pub use sort::sort_by;
diff --git a/src/compute/sort/primitive/sort.rs b/src/compute/sort/primitive/sort.rs
index 224a9fd9bc..197ea29941 100644
--- a/src/compute/sort/primitive/sort.rs
+++ b/src/compute/sort/primitive/sort.rs
@@ -15,7 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::buffer::MutableBuffer;
+use crate::bitmap::Bitmap;
+use crate::buffer::{Buffer, MutableBuffer};
 use crate::{
     array::{Array, PrimitiveArray},
     bitmap::{utils::SlicesIterator, MutableBitmap},
@@ -24,11 +25,32 @@ use crate::{
 
 use super::super::SortOptions;
 
-fn sort_inner<T, F>(values: &mut [T], mut cmp: F, descending: bool)
+/// # Safety
+/// `indices[i] < values.len()` for all i
+#[inline]
+fn k_element_sort_inner<T, F>(values: &mut [T], descending: bool, limit: usize, mut cmp: F)
 where
     T: NativeType,
     F: FnMut(&T, &T) -> std::cmp::Ordering,
 {
+    if descending {
+        let (before, _, _) = values.select_nth_unstable_by(limit, |x, y| cmp(x, y).reverse());
+        before.sort_unstable_by(|x, y| cmp(x, y));
+    } else {
+        let (before, _, _) = values.select_nth_unstable_by(limit, |x, y| cmp(x, y));
+        before.sort_unstable_by(|x, y| cmp(x, y));
+    }
+}
+
+fn sort_values<T, F>(values: &mut [T], mut cmp: F, descending: bool, limit: usize)
+where
+    T: NativeType,
+    F: FnMut(&T, &T) -> std::cmp::Ordering,
+{
+    if limit != values.len() {
+        return k_element_sort_inner(values, descending, limit, cmp);
+    }
+
     if descending {
         values.sort_unstable_by(|x, y| cmp(x, y).reverse());
     } else {
@@ -36,58 +58,106 @@ where
     };
 }
 
+fn sort_nullable<T, F>(
+    values: &[T],
+    validity: &Bitmap,
+    cmp: F,
+    options: &SortOptions,
+    limit: usize,
+) -> (Buffer<T>, Option<Bitmap>)
+where
+    T: NativeType,
+    F: FnMut(&T, &T) -> std::cmp::Ordering,
+{
+    assert!(limit <= values.len());
+    if options.nulls_first && limit < validity.null_count() {
+        let mut buffer = MutableBuffer::<T>::with_capacity(limit);
+        buffer.extend_constant(limit, T::default());
+        let bitmap = MutableBitmap::from_trusted_len_iter(std::iter::repeat(false).take(limit));
+        return (buffer.into(), bitmap.into());
+    }
+
+    let nulls = std::iter::repeat(false).take(validity.null_count());
+    let valids = std::iter::repeat(true).take(values.len() - validity.null_count());
+
+    let mut buffer = MutableBuffer::<T>::with_capacity(values.len());
+    let mut new_validity = MutableBitmap::with_capacity(values.len());
+    let slices = SlicesIterator::new(validity);
+
+    if options.nulls_first {
+        // validity is [0,0,0,...,1,1,1,1]
+        new_validity.extend_from_trusted_len_iter(nulls.chain(valids).take(limit));
+
+        // extend buffer with constants followed by non-null values
+        buffer.extend_constant(validity.null_count(), T::default());
+        for (start, len) in slices {
+            buffer.extend_from_slice(&values[start..start + len])
+        }
+
+        // sort values
+        sort_values(
+            &mut buffer.as_mut_slice()[validity.null_count()..],
+            cmp,
+            options.descending,
+            limit - validity.null_count(),
+        );
+    } else {
+        // validity is [1,1,1,...,0,0,0,0]
+        new_validity.extend_from_trusted_len_iter(valids.chain(nulls).take(limit));
+
+        // extend buffer with non-null values
+        for (start, len) in slices {
+            buffer.extend_from_slice(&values[start..start + len])
+        }
+
+        // sort all non-null values
+        sort_values(
+            &mut buffer.as_mut_slice(),
+            cmp,
+            options.descending,
+            limit - validity.null_count(),
+        );
+
+        if limit > values.len() - validity.null_count() {
+            // extend remaining with nulls
+            buffer.extend_constant(validity.null_count(), T::default());
+        }
+    };
+    // values are sorted, we can now truncate the remaining.
+    buffer.truncate(limit);
+
+    (buffer.into(), new_validity.into())
+}
+
 /// Sorts a [`PrimitiveArray`] according to `cmp` comparator and [`SortOptions`].
-pub fn sort_by<T, F>(array: &PrimitiveArray<T>, cmp: F, options: &SortOptions) -> PrimitiveArray<T>
+pub fn sort_by<T, F>(
+    array: &PrimitiveArray<T>,
+    cmp: F,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> PrimitiveArray<T>
 where
     T: NativeType,
     F: FnMut(&T, &T) -> std::cmp::Ordering,
 {
+    let limit = limit.unwrap_or_else(|| array.len());
+    let limit = limit.min(array.len());
+
     let values = array.values();
     let validity = array.validity();
 
     let (buffer, validity) = if let Some(validity) = validity {
-        let nulls = (0..validity.null_count()).map(|_| false);
-        let valids = (validity.null_count()..array.len()).map(|_| true);
-
-        let mut buffer = MutableBuffer::<T>::with_capacity(array.len());
-        let mut new_validity = MutableBitmap::with_capacity(array.len());
-        let slices = SlicesIterator::new(validity);
-
-        if options.nulls_first {
-            nulls
-                .chain(valids)
-                .for_each(|value| unsafe { new_validity.push_unchecked(value) });
-            (0..validity.null_count()).for_each(|_| buffer.push(T::default()));
-            for (start, len) in slices {
-                buffer.extend_from_slice(&values[start..start + len])
-            }
-            sort_inner(
-                &mut buffer.as_mut_slice()[validity.null_count()..],
-                cmp,
-                options.descending,
-            )
-        } else {
-            valids
-                .chain(nulls)
-                .for_each(|value| unsafe { new_validity.push_unchecked(value) });
-            for (start, len) in slices {
-                buffer.extend_from_slice(&values[start..start + len])
-            }
-            sort_inner(&mut buffer.as_mut_slice(), cmp, options.descending);
-
-            (0..validity.null_count()).for_each(|_| buffer.push(T::default()));
-        };
-
-        (buffer, new_validity.into())
+        sort_nullable(values, validity, cmp, options, limit)
     } else {
         let mut buffer = MutableBuffer::<T>::new();
         buffer.extend_from_slice(values);
 
-        sort_inner(&mut buffer.as_mut_slice(), cmp, options.descending);
+        sort_values(&mut buffer.as_mut_slice(), cmp, options.descending, limit);
+        buffer.truncate(limit);
 
-        (buffer, None)
+        (buffer.into(), None)
     };
-    PrimitiveArray::<T>::from_data(array.data_type().clone(), buffer.into(), validity)
+    PrimitiveArray::<T>::from_data(array.data_type().clone(), buffer, validity)
 }
 
 #[cfg(test)]
@@ -107,8 +177,13 @@ mod tests {
         T: NativeType + std::cmp::Ord,
     {
         let input = PrimitiveArray::<T>::from(data).to(data_type.clone());
-        let expected = PrimitiveArray::<T>::from(expected_data).to(data_type);
-        let output = sort_by(&input, ord::total_cmp, &options);
+        let expected = PrimitiveArray::<T>::from(expected_data).to(data_type.clone());
+        let output = sort_by(&input, ord::total_cmp, &options, None);
+        assert_eq!(expected, output);
+
+        // with limit
+        let expected = PrimitiveArray::<T>::from(&expected_data[..3]).to(data_type);
+        let output = sort_by(&input, ord::total_cmp, &options, Some(3));
         assert_eq!(expected, output)
     }
 
diff --git a/src/compute/sort/utf8.rs b/src/compute/sort/utf8.rs
new file mode 100644
index 0000000000..20f2d65fea
--- /dev/null
+++ b/src/compute/sort/utf8.rs
@@ -0,0 +1,37 @@
+use crate::array::{Array, Index, Offset, PrimitiveArray, Utf8Array};
+use crate::array::{DictionaryArray, DictionaryKey};
+
+use super::common;
+use super::SortOptions;
+
+pub(super) fn indices_sorted_unstable_by<I: Index, O: Offset>(
+    array: &Utf8Array<O>,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> PrimitiveArray<I> {
+    let get = |idx| unsafe { array.value_unchecked(idx as usize) };
+    let cmp = |lhs: &&str, rhs: &&str| lhs.cmp(rhs);
+    common::indices_sorted_unstable_by(array.validity(), get, cmp, array.len(), options, limit)
+}
+
+pub(super) fn indices_sorted_unstable_by_dictionary<I: Index, K: DictionaryKey, O: Offset>(
+    array: &DictionaryArray<K>,
+    options: &SortOptions,
+    limit: Option<usize>,
+) -> PrimitiveArray<I> {
+    let keys = array.keys();
+
+    let dict = array
+        .values()
+        .as_any()
+        .downcast_ref::<Utf8Array<O>>()
+        .unwrap();
+
+    let get = |idx| unsafe {
+        let index = keys.value_unchecked(idx as usize);
+        // Note: there is no check that the keys are within bounds of the dictionary.
+        dict.value(index.to_usize().unwrap())
+    };
+    let cmp = |lhs: &&str, rhs: &&str| lhs.cmp(rhs);
+    common::indices_sorted_unstable_by(array.validity(), get, cmp, array.len(), options, limit)
+}
diff --git a/src/compute/take/generic_binary.rs b/src/compute/take/generic_binary.rs
index 3d418aaeb0..4d9af24bb3 100644
--- a/src/compute/take/generic_binary.rs
+++ b/src/compute/take/generic_binary.rs
@@ -14,7 +14,6 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
-
 use crate::{
     array::{Array, GenericBinaryArray, Offset, PrimitiveArray},
     bitmap::{Bitmap, MutableBitmap},
diff --git a/src/compute/take/mod.rs b/src/compute/take/mod.rs
index 06dc7e990b..1517f37e40 100644
--- a/src/compute/take/mod.rs
+++ b/src/compute/take/mod.rs
@@ -21,6 +21,7 @@ use crate::{
     array::{new_empty_array, Array, NullArray, PrimitiveArray},
     datatypes::{DataType, IntervalUnit},
     error::Result,
+    types::days_ms,
 };
 
 pub use crate::array::Index;
@@ -76,6 +77,7 @@ pub fn take<O: Index>(values: &dyn Array, indices: &PrimitiveArray<O>) -> Result
         | DataType::Time64(_)
         | DataType::Duration(_)
         | DataType::Timestamp(_, _) => downcast_take!(i64, values, indices),
+        DataType::Interval(IntervalUnit::DayTime) => downcast_take!(days_ms, values, indices),
         DataType::UInt8 => downcast_take!(u8, values, indices),
         DataType::UInt16 => downcast_take!(u16, values, indices),
         DataType::UInt32 => downcast_take!(u32, values, indices),
@@ -146,7 +148,7 @@ pub fn can_take(data_type: &DataType) -> bool {
         | DataType::Int32
         | DataType::Date32
         | DataType::Time32(_)
-        | DataType::Interval(IntervalUnit::YearMonth)
+        | DataType::Interval(_)
         | DataType::Int64
         | DataType::Date64
         | DataType::Time64(_)
@@ -318,6 +320,8 @@ mod tests {
             Timestamp(TimeUnit::Nanosecond, None),
             Time64(TimeUnit::Microsecond),
             Time64(TimeUnit::Nanosecond),
+            Interval(IntervalUnit::DayTime),
+            Interval(IntervalUnit::YearMonth),
             Date32,
             Time32(TimeUnit::Second),
             Time32(TimeUnit::Millisecond),
@@ -334,11 +338,10 @@ mod tests {
 
         datatypes.into_iter().for_each(|d1| {
             let array = new_null_array(d1.clone(), 10);
+            let indices = Int32Array::from(&[Some(1), Some(2), None, Some(3)]);
             if can_take(&d1) {
-                let indices = Int32Array::from(&[Some(1), Some(2), None, Some(3)]);
                 assert!(take(array.as_ref(), &indices).is_ok());
             } else {
-                let indices = Int32Array::from(&[Some(1), Some(2), None, Some(3)]);
                 assert!(take(array.as_ref(), &indices).is_err());
             }
         });
diff --git a/src/compute/take/primitive.rs b/src/compute/take/primitive.rs
index cff907a015..8fdba3b4ad 100644
--- a/src/compute/take/primitive.rs
+++ b/src/compute/take/primitive.rs
@@ -14,7 +14,6 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
-
 use crate::{
     array::{Array, PrimitiveArray},
     bitmap::{Bitmap, MutableBitmap},
diff --git a/src/datatypes/field.rs b/src/datatypes/field.rs
index b83bb960b0..205a9a041d 100644
--- a/src/datatypes/field.rs
+++ b/src/datatypes/field.rs
@@ -187,7 +187,7 @@ impl Field {
                                 continue;
                             }
                             is_new_field = false;
-                            self_field.try_merge(&from_field)?;
+                            self_field.try_merge(from_field)?;
                         }
                         if is_new_field {
                             nested_fields.push(from_field.clone());
diff --git a/src/ffi/array.rs b/src/ffi/array.rs
index 65f708cf2e..0929aaebc9 100644
--- a/src/ffi/array.rs
+++ b/src/ffi/array.rs
@@ -81,6 +81,7 @@ pub fn try_from<A: ArrowArrayRef>(array: A) -> Result<Box<dyn Array>> {
 mod tests {
     use super::*;
     use crate::array::*;
+    use crate::datatypes::TimeUnit;
     use crate::{error::Result, ffi};
     use std::sync::Arc;
 
@@ -146,6 +147,15 @@ mod tests {
         test_round_trip(data)
     }
 
+    #[test]
+    fn test_timestamp_tz() -> Result<()> {
+        let data = Int64Array::from(&vec![Some(2), None, None]).to(DataType::Timestamp(
+            TimeUnit::Second,
+            Some("UTC".to_string()),
+        ));
+        test_round_trip(data)
+    }
+
     #[test]
     fn test_large_binary() -> Result<()> {
         let data =
@@ -168,4 +178,25 @@ mod tests {
 
         test_round_trip(array)
     }
+
+    #[test]
+    fn test_list_list() -> Result<()> {
+        let data = vec![
+            Some(vec![
+                Some(vec![None]),
+                Some(vec![Some(2)]),
+                Some(vec![Some(3)]),
+            ]),
+            None,
+            Some(vec![Some(vec![Some(4), None, Some(6)])]),
+        ];
+
+        let mut array =
+            MutableListArray::<i32, MutableListArray<i32, MutablePrimitiveArray<i32>>>::new();
+        array.try_extend(data)?;
+
+        let array: ListArray<i32> = array.into();
+
+        test_round_trip(array)
+    }
 }
diff --git a/src/ffi/ffi.rs b/src/ffi/ffi.rs
index 725dc42cbe..f4c1f77669 100644
--- a/src/ffi/ffi.rs
+++ b/src/ffi/ffi.rs
@@ -29,7 +29,7 @@ use crate::{
         bytes::{Bytes, Deallocation},
         Buffer,
     },
-    datatypes::{DataType, Field, TimeUnit},
+    datatypes::{DataType, Field, IntervalUnit, TimeUnit},
     error::{ArrowError, Result},
     types::NativeType,
 };
@@ -200,6 +200,12 @@ fn to_field(schema: &Ffi_ArrowSchema) -> Result<Field> {
         "ttm" => DataType::Time32(TimeUnit::Millisecond),
         "ttu" => DataType::Time64(TimeUnit::Microsecond),
         "ttn" => DataType::Time64(TimeUnit::Nanosecond),
+        "tDs" => DataType::Duration(TimeUnit::Second),
+        "tDm" => DataType::Duration(TimeUnit::Millisecond),
+        "tDu" => DataType::Duration(TimeUnit::Microsecond),
+        "tDn" => DataType::Duration(TimeUnit::Nanosecond),
+        "tiM" => DataType::Interval(IntervalUnit::YearMonth),
+        "tiD" => DataType::Interval(IntervalUnit::DayTime),
         "+l" => {
             let child = schema.child(0);
             DataType::List(Box::new(to_field(child)?))
@@ -215,10 +221,43 @@ fn to_field(schema: &Ffi_ArrowSchema) -> Result<Field> {
             DataType::Struct(children)
         }
         other => {
-            return Err(ArrowError::Ffi(format!(
-                "The datatype \"{}\" is still not supported in Rust implementation",
-                other
-            )))
+            let parts = other.split(':').collect::<Vec<_>>();
+            if parts.len() == 2 && parts[0] == "tss" {
+                DataType::Timestamp(TimeUnit::Second, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "tsm" {
+                DataType::Timestamp(TimeUnit::Millisecond, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "tsu" {
+                DataType::Timestamp(TimeUnit::Microsecond, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "tsn" {
+                DataType::Timestamp(TimeUnit::Nanosecond, Some(parts[1].to_string()))
+            } else if parts.len() == 2 && parts[0] == "d" {
+                let parts = parts[1].split(',').collect::<Vec<_>>();
+                if parts.len() < 2 || parts.len() > 3 {
+                    return Err(ArrowError::Ffi(
+                        "Decimal must contain 2 or 3 comma-separated values".to_string(),
+                    ));
+                };
+                if parts.len() == 3 {
+                    let bit_width = parts[0].parse::<usize>().map_err(|_| {
+                        ArrowError::Ffi("Decimal bit width is not a valid integer".to_string())
+                    })?;
+                    if bit_width != 128 {
+                        return Err(ArrowError::Ffi("Decimal256 is not supported".to_string()));
+                    }
+                }
+                let precision = parts[0].parse::<usize>().map_err(|_| {
+                    ArrowError::Ffi("Decimal precision is not a valid integer".to_string())
+                })?;
+                let scale = parts[1].parse::<usize>().map_err(|_| {
+                    ArrowError::Ffi("Decimal scale is not a valid integer".to_string())
+                })?;
+                DataType::Decimal(precision, scale)
+            } else {
+                return Err(ArrowError::Ffi(format!(
+                    "The datatype \"{}\" is still not supported in Rust implementation",
+                    other
+                )));
+            }
         }
     };
     Ok(Field::new(schema.name(), data_type, schema.nullable()))
@@ -250,15 +289,33 @@ fn to_format(data_type: &DataType) -> Result<String> {
         DataType::Time32(TimeUnit::Millisecond) => "ttm",
         DataType::Time64(TimeUnit::Microsecond) => "ttu",
         DataType::Time64(TimeUnit::Nanosecond) => "ttn",
+        DataType::Duration(TimeUnit::Second) => "tDs",
+        DataType::Duration(TimeUnit::Millisecond) => "tDm",
+        DataType::Duration(TimeUnit::Microsecond) => "tDu",
+        DataType::Duration(TimeUnit::Nanosecond) => "tDn",
+        DataType::Interval(IntervalUnit::YearMonth) => "tiM",
+        DataType::Interval(IntervalUnit::DayTime) => "tiD",
+        DataType::Timestamp(unit, tz) => {
+            let unit = match unit {
+                TimeUnit::Second => "s",
+                TimeUnit::Millisecond => "m",
+                TimeUnit::Microsecond => "u",
+                TimeUnit::Nanosecond => "n",
+            };
+            return Ok(format!(
+                "ts{}:{}",
+                unit,
+                tz.as_ref().map(|x| x.as_ref()).unwrap_or("")
+            ));
+        }
+        DataType::Decimal(precision, scale) => return Ok(format!("d:{},{}", precision, scale)),
         DataType::List(_) => "+l",
         DataType::LargeList(_) => "+L",
         DataType::Struct(_) => "+s",
-        z => {
-            return Err(ArrowError::Ffi(format!(
-                "The datatype \"{:?}\" is still not supported in Rust implementation",
-                z
-            )))
-        }
+        DataType::FixedSizeBinary(size) => return Ok(format!("w{}", size)),
+        DataType::FixedSizeList(_, size) => return Ok(format!("+w:{}", size)),
+        DataType::Union(_) => todo!(),
+        _ => todo!(),
     }
     .to_string())
 }
@@ -607,7 +664,7 @@ impl ArrowArrayRef for Arc<ArrowArray> {
     }
 
     fn parent(&self) -> &Arc<ArrowArray> {
-        &self
+        self
     }
 
     fn array(&self) -> &Ffi_ArrowArray {
@@ -643,10 +700,6 @@ impl<'a> ArrowArrayRef for ArrowArrayChild<'a> {
     fn schema(&self) -> &Ffi_ArrowSchema {
         self.schema
     }
-
-    fn child(&self, _: usize) -> Result<ArrowArrayChild> {
-        todo!()
-    }
 }
 
 impl<'a> ArrowArrayChild<'a> {
diff --git a/src/io/csv/read/deserialize.rs b/src/io/csv/read/deserialize.rs
index 351674ead7..84fc1bead6 100644
--- a/src/io/csv/read/deserialize.rs
+++ b/src/io/csv/read/deserialize.rs
@@ -174,7 +174,7 @@ where
     F: Fn(&[ByteRecord], usize, DataType, usize) -> Result<Arc<dyn Array>>,
 {
     let projection: Vec<usize> = match projection {
-        Some(ref v) => v.to_vec(),
+        Some(v) => v.to_vec(),
         None => fields.iter().enumerate().map(|(i, _)| i).collect(),
     };
     let projected_fields: Vec<Field> = projection.iter().map(|i| fields[*i].clone()).collect();
diff --git a/src/io/csv/read/infer_schema.rs b/src/io/csv/read/infer_schema.rs
index b52f1bf721..be48e3ef67 100644
--- a/src/io/csv/read/infer_schema.rs
+++ b/src/io/csv/read/infer_schema.rs
@@ -69,7 +69,7 @@ pub fn infer_schema<R: Read + Seek, F: Fn(&str) -> DataType>(
         match possibilities.len() {
             1 => {
                 for dtype in possibilities.iter() {
-                    fields.push(Field::new(&field_name, dtype.clone(), true));
+                    fields.push(Field::new(field_name, dtype.clone(), true));
                 }
             }
             2 => {
@@ -77,13 +77,13 @@ pub fn infer_schema<R: Read + Seek, F: Fn(&str) -> DataType>(
                     && possibilities.contains(&DataType::Float64)
                 {
                     // we have an integer and double, fall down to double
-                    fields.push(Field::new(&field_name, DataType::Float64, true));
+                    fields.push(Field::new(field_name, DataType::Float64, true));
                 } else {
                     // default to Utf8 for conflicting datatypes (e.g bool and int)
-                    fields.push(Field::new(&field_name, DataType::Utf8, true));
+                    fields.push(Field::new(field_name, DataType::Utf8, true));
                 }
             }
-            _ => fields.push(Field::new(&field_name, DataType::Utf8, true)),
+            _ => fields.push(Field::new(field_name, DataType::Utf8, true)),
         }
     }
 
diff --git a/src/io/ipc/read/deserialize.rs b/src/io/ipc/read/deserialize.rs
index a8105270f5..8393a668c9 100644
--- a/src/io/ipc/read/deserialize.rs
+++ b/src/io/ipc/read/deserialize.rs
@@ -303,7 +303,7 @@ where
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -336,7 +336,7 @@ fn read_boolean<R: Read + Seek>(
     let length = field_node.length() as usize;
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -371,7 +371,7 @@ where
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -417,7 +417,7 @@ where
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -461,7 +461,7 @@ fn read_fixed_size_binary<R: Read + Seek>(
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -502,7 +502,7 @@ where
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -549,7 +549,7 @@ fn read_fixed_size_list<R: Read + Seek>(
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
@@ -585,7 +585,7 @@ fn read_struct<R: Read + Seek>(
 
     let validity = read_validity(
         buffers,
-        &field_node,
+        field_node,
         reader,
         block_offset,
         is_little_endian,
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index 870de58f47..8860426bae 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -115,12 +115,10 @@ pub fn read_file_metadata<R: Read + Seek>(reader: &mut R) -> Result<FileMetadata
         let mut message_size: [u8; 4] = [0; 4];
         reader.seek(SeekFrom::Start(block.offset() as u64))?;
         reader.read_exact(&mut message_size)?;
-        let footer_len = if message_size == CONTINUATION_MARKER {
+        if message_size == CONTINUATION_MARKER {
             reader.read_exact(&mut message_size)?;
-            i32::from_le_bytes(message_size)
-        } else {
-            i32::from_le_bytes(message_size)
         };
+        let footer_len = i32::from_le_bytes(message_size);
 
         let mut block_data = vec![0; footer_len as usize];
 
diff --git a/src/io/ipc/read/stream.rs b/src/io/ipc/read/stream.rs
index 73536c2d53..fe40bf2233 100644
--- a/src/io/ipc/read/stream.rs
+++ b/src/io/ipc/read/stream.rs
@@ -132,7 +132,7 @@ pub fn read_next<R: Read>(
                 batch,
                 metadata.schema.clone(),
                 metadata.is_little_endian,
-                &dictionaries_by_field,
+                dictionaries_by_field,
                 &mut reader,
                 0,
             )
diff --git a/src/io/ipc/write/common.rs b/src/io/ipc/write/common.rs
index 2b54e4a98d..fd7f820379 100644
--- a/src/io/ipc/write/common.rs
+++ b/src/io/ipc/write/common.rs
@@ -386,7 +386,7 @@ pub fn write_message<W: Write>(
     let aligned_size = (flatbuf_size + prefix_size + a) & !a;
     let padding_bytes = aligned_size - flatbuf_size - prefix_size;
 
-    write_continuation(writer, &write_options, (aligned_size - prefix_size) as i32)?;
+    write_continuation(writer, write_options, (aligned_size - prefix_size) as i32)?;
 
     // write the flatbuf
     if flatbuf_size > 0 {
diff --git a/src/io/json/read/reader.rs b/src/io/json/read/reader.rs
index 69c1ed721a..c78fb92e2e 100644
--- a/src/io/json/read/reader.rs
+++ b/src/io/json/read/reader.rs
@@ -341,8 +341,8 @@ mod tests {
             .as_any()
             .downcast_ref::<BooleanArray>()
             .unwrap();
-        assert_eq!(false, cc.value(0));
-        assert_eq!(true, cc.value(10));
+        assert!(!cc.value(0));
+        assert!(cc.value(10));
         let dd = batch
             .column(d.0)
             .as_any()
@@ -381,34 +381,34 @@ mod tests {
             .as_any()
             .downcast_ref::<PrimitiveArray<i64>>()
             .unwrap();
-        assert_eq!(true, aa.is_valid(0));
-        assert_eq!(false, aa.is_valid(1));
-        assert_eq!(false, aa.is_valid(11));
+        assert!(aa.is_valid(0));
+        assert!(!aa.is_valid(1));
+        assert!(!aa.is_valid(11));
         let bb = batch
             .column(b.0)
             .as_any()
             .downcast_ref::<PrimitiveArray<f64>>()
             .unwrap();
-        assert_eq!(true, bb.is_valid(0));
-        assert_eq!(false, bb.is_valid(2));
-        assert_eq!(false, bb.is_valid(11));
+        assert!(bb.is_valid(0));
+        assert!(!bb.is_valid(2));
+        assert!(!bb.is_valid(11));
         let cc = batch
             .column(c.0)
             .as_any()
             .downcast_ref::<BooleanArray>()
             .unwrap();
-        assert_eq!(true, cc.is_valid(0));
-        assert_eq!(false, cc.is_valid(4));
-        assert_eq!(false, cc.is_valid(11));
+        assert!(cc.is_valid(0));
+        assert!(!cc.is_valid(4));
+        assert!(!cc.is_valid(11));
         let dd = batch
             .column(d.0)
             .as_any()
             .downcast_ref::<Utf8Array<i32>>()
             .unwrap();
-        assert_eq!(false, dd.is_valid(0));
-        assert_eq!(true, dd.is_valid(1));
-        assert_eq!(false, dd.is_valid(4));
-        assert_eq!(false, dd.is_valid(11));
+        assert!(!dd.is_valid(0));
+        assert!(dd.is_valid(1));
+        assert!(!dd.is_valid(4));
+        assert!(!dd.is_valid(11));
     }
 
     #[test]
@@ -451,7 +451,7 @@ mod tests {
             .unwrap();
         assert_eq!(1, aa.value(0));
         // test that a 64bit value is returned as null due to overflowing
-        assert_eq!(false, aa.is_valid(11));
+        assert!(!aa.is_valid(11));
         let bb = batch
             .column(b.0)
             .as_any()
@@ -546,7 +546,7 @@ mod tests {
         assert_eq!(9, bb.len());
         assert!((2.0 - bb.value(0)).abs() < f64::EPSILON);
         assert!((-6.1 - bb.value(5)).abs() < f64::EPSILON);
-        assert_eq!(false, bb.is_valid(7));
+        assert!(!bb.is_valid(7));
 
         let cc = batch
             .column(c.0)
@@ -556,9 +556,9 @@ mod tests {
         let cc = cc.values();
         let cc = cc.as_any().downcast_ref::<BooleanArray>().unwrap();
         assert_eq!(6, cc.len());
-        assert_eq!(false, cc.value(0));
-        assert_eq!(false, cc.value(4));
-        assert_eq!(false, cc.is_valid(5));
+        assert!(!cc.value(0));
+        assert!(!cc.value(4));
+        assert!(!cc.is_valid(5));
     }
 
     #[test]
diff --git a/src/io/json_integration/read.rs b/src/io/json_integration/read.rs
index 266c35aa2d..6063335a95 100644
--- a/src/io/json_integration/read.rs
+++ b/src/io/json_integration/read.rs
@@ -181,7 +181,7 @@ fn to_list<O: Offset>(
 
     let child_field = ListArray::<O>::get_child_field(&data_type);
     let children = &json_col.children.as_ref().unwrap()[0];
-    let values = to_array(&child_field, children, dictionaries)?;
+    let values = to_array(child_field, children, dictionaries)?;
     let offsets = to_offsets::<O>(json_col.offset.as_ref());
     Ok(Arc::new(ListArray::<O>::from_data(
         data_type, offsets, values, validity,
@@ -290,7 +290,7 @@ pub fn to_array(
             let validity = to_validity(&json_col.validity);
 
             let children = &json_col.children.as_ref().unwrap()[0];
-            let values = to_array(&child_field, children, dictionaries)?;
+            let values = to_array(child_field, children, dictionaries)?;
 
             Ok(Arc::new(FixedSizeListArray::from_data(
                 data_type.clone(),
@@ -337,7 +337,7 @@ pub fn to_record_batch(
         .fields()
         .iter()
         .zip(&json_batch.columns)
-        .map(|(field, json_col)| to_array(field, &json_col, json_dictionaries))
+        .map(|(field, json_col)| to_array(field, json_col, json_dictionaries))
         .collect::<Result<Vec<_>>>()?;
 
     RecordBatch::try_new(Arc::new(schema.clone()), columns)
diff --git a/src/io/parquet/mod.rs b/src/io/parquet/mod.rs
index def7c45e52..1150fec4db 100644
--- a/src/io/parquet/mod.rs
+++ b/src/io/parquet/mod.rs
@@ -419,7 +419,7 @@ mod tests_integration {
             version: Version::V1,
         };
 
-        let parquet_schema = to_parquet_schema(&schema)?;
+        let parquet_schema = to_parquet_schema(schema)?;
         let descritors = parquet_schema.columns().to_vec().into_iter();
 
         let row_groups = batches.iter().map(|batch| {
diff --git a/src/io/parquet/read/binary/basic.rs b/src/io/parquet/read/binary/basic.rs
index 2ca0ec03e1..535741798a 100644
--- a/src/io/parquet/read/binary/basic.rs
+++ b/src/io/parquet/read/binary/basic.rs
@@ -35,14 +35,14 @@ fn read_dict_buffer<O: Offset>(
     let bit_width = indices_buffer[0];
     let indices_buffer = &indices_buffer[1..];
 
-    let (_, consumed) = uleb128::decode(&indices_buffer);
+    let (_, consumed) = uleb128::decode(indices_buffer);
     let indices_buffer = &indices_buffer[consumed..];
 
     let non_null_indices_len = indices_buffer.len() * 8 / bit_width as usize;
 
     let mut indices = bitpacking::Decoder::new(indices_buffer, bit_width, non_null_indices_len);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     for run in validity_iterator {
         match run {
@@ -96,7 +96,7 @@ fn read_delta_optional<O: Offset>(
     // values_buffer: first 4 bytes are len, remaining is values
     let mut values_iterator = delta_length_byte_array::Decoder::new(values_buffer);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     // offsets:
     for run in validity_iterator {
@@ -148,7 +148,7 @@ fn read_plain_optional<O: Offset>(
     // values_buffer: first 4 bytes are len, remaining is values
     let mut values_iterator = utils::BinaryIter::new(values_buffer);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     for run in validity_iterator {
         match run {
diff --git a/src/io/parquet/read/boolean/basic.rs b/src/io/parquet/read/boolean/basic.rs
index 06d3caf548..29f959e49e 100644
--- a/src/io/parquet/read/boolean/basic.rs
+++ b/src/io/parquet/read/boolean/basic.rs
@@ -23,7 +23,7 @@ fn read_optional(
     values: &mut MutableBitmap,
     validity: &mut MutableBitmap,
 ) {
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     // in PLAIN, booleans are LSB bitpacked and thus we can read them as if they were a bitmap.
     // note that `values_buffer` contains only non-null values.
diff --git a/src/io/parquet/read/fixed_size_binary.rs b/src/io/parquet/read/fixed_size_binary.rs
index 9356d49395..9408aaa7d2 100644
--- a/src/io/parquet/read/fixed_size_binary.rs
+++ b/src/io/parquet/read/fixed_size_binary.rs
@@ -34,7 +34,7 @@ pub(crate) fn read_dict_buffer(
     let bit_width = indices_buffer[0];
     let indices_buffer = &indices_buffer[1..];
 
-    let (_, consumed) = uleb128::decode(&indices_buffer);
+    let (_, consumed) = uleb128::decode(indices_buffer);
     let indices_buffer = &indices_buffer[consumed..];
 
     let non_null_indices_len = (indices_buffer.len() * 8 / bit_width as usize) as u32;
@@ -42,7 +42,7 @@ pub(crate) fn read_dict_buffer(
     let mut indices =
         bitpacking::Decoder::new(indices_buffer, bit_width, non_null_indices_len as usize);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     for run in validity_iterator {
         match run {
@@ -89,7 +89,7 @@ pub(crate) fn read_optional(
     assert_eq!(values_buffer.len() % size, 0);
     let mut values_iterator = values_buffer.chunks_exact(size);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     for run in validity_iterator {
         match run {
diff --git a/src/io/parquet/read/primitive/basic.rs b/src/io/parquet/read/primitive/basic.rs
index d136874c99..9cfaa7ced2 100644
--- a/src/io/parquet/read/primitive/basic.rs
+++ b/src/io/parquet/read/primitive/basic.rs
@@ -34,14 +34,14 @@ fn read_dict_buffer_optional<T, A, F>(
     let bit_width = indices_buffer[0];
     let indices_buffer = &indices_buffer[1..];
 
-    let (_, consumed) = uleb128::decode(&indices_buffer);
+    let (_, consumed) = uleb128::decode(indices_buffer);
     let indices_buffer = &indices_buffer[consumed..];
 
     let non_null_indices_len = indices_buffer.len() * 8 / bit_width as usize;
 
     let mut indices = bitpacking::Decoder::new(indices_buffer, bit_width, non_null_indices_len);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     for run in validity_iterator {
         match run {
@@ -89,7 +89,7 @@ fn read_nullable<T, A, F>(
 {
     let mut chunks = ExactChunksIter::<T>::new(values_buffer);
 
-    let validity_iterator = hybrid_rle::Decoder::new(&validity_buffer, 1);
+    let validity_iterator = hybrid_rle::Decoder::new(validity_buffer, 1);
 
     for run in validity_iterator {
         match run {
diff --git a/src/io/parquet/read/statistics/binary.rs b/src/io/parquet/read/statistics/binary.rs
index 682b7572b2..4f883fa6e1 100644
--- a/src/io/parquet/read/statistics/binary.rs
+++ b/src/io/parquet/read/statistics/binary.rs
@@ -57,12 +57,12 @@ impl TryFrom<&ParquetByteArrayStatistics> for Utf8Statistics {
             min_value: stats
                 .min_value
                 .as_ref()
-                .map(|x| std::str::from_utf8(&x).map(|x| x.to_string()))
+                .map(|x| std::str::from_utf8(x).map(|x| x.to_string()))
                 .transpose()?,
             max_value: stats
                 .max_value
                 .as_ref()
-                .map(|x| std::str::from_utf8(&x).map(|x| x.to_string()))
+                .map(|x| std::str::from_utf8(x).map(|x| x.to_string()))
                 .transpose()?,
         })
     }
diff --git a/src/io/parquet/write/levels.rs b/src/io/parquet/write/levels.rs
index f81422bced..59b74b80d8 100644
--- a/src/io/parquet/write/levels.rs
+++ b/src/io/parquet/write/levels.rs
@@ -164,7 +164,7 @@ impl<'a, O: Offset> NestedInfo<'a, O> {
     }
 
     pub fn offsets(&self) -> &'a [O] {
-        &self.offsets
+        self.offsets
     }
 }
 
@@ -197,13 +197,13 @@ pub fn write_rep_levels<O: Offset>(
     match version {
         Version::V1 => {
             write_levels_v1(buffer, |buffer: &mut Vec<u8>| {
-                let levels = RepLevelsIter::new(&nested.offsets);
+                let levels = RepLevelsIter::new(nested.offsets);
                 encode_u32(buffer, levels, num_bits)?;
                 Ok(())
             })?;
         }
         Version::V2 => {
-            let levels = RepLevelsIter::new(&nested.offsets);
+            let levels = RepLevelsIter::new(nested.offsets);
 
             encode_u32(buffer, levels, num_bits)?;
         }
@@ -224,13 +224,13 @@ pub fn write_def_levels<O: Offset>(
     match version {
         Version::V1 => {
             write_levels_v1(buffer, |buffer: &mut Vec<u8>| {
-                let levels = DefLevelsIter::new(&nested.offsets, &nested.validity, validity);
+                let levels = DefLevelsIter::new(nested.offsets, nested.validity, validity);
                 encode_u32(buffer, levels, num_bits)?;
                 Ok(())
             })?;
         }
         Version::V2 => {
-            let levels = DefLevelsIter::new(&nested.offsets, &nested.validity, validity);
+            let levels = DefLevelsIter::new(nested.offsets, nested.validity, validity);
             encode_u32(buffer, levels, num_bits)?;
         }
     }
diff --git a/src/io/parquet/write/mod.rs b/src/io/parquet/write/mod.rs
index deabba2742..612c9f7b0f 100644
--- a/src/io/parquet/write/mod.rs
+++ b/src/io/parquet/write/mod.rs
@@ -248,11 +248,11 @@ pub fn array_to_page(
         ),
         DataType::Decimal(precision, _) => {
             let precision = *precision;
+            let array = array
+                .as_any()
+                .downcast_ref::<PrimitiveArray<i128>>()
+                .unwrap();
             if precision <= 9 {
-                let array = array
-                    .as_any()
-                    .downcast_ref::<PrimitiveArray<i128>>()
-                    .unwrap();
                 let values = array.values().iter().map(|x| *x as i32);
                 let values = Buffer::from_trusted_len_iter(values);
                 let array = PrimitiveArray::<i32>::from_data(
@@ -262,10 +262,6 @@ pub fn array_to_page(
                 );
                 primitive::array_to_page::<i32, i32>(&array, options, descriptor)
             } else if precision <= 18 {
-                let array = array
-                    .as_any()
-                    .downcast_ref::<PrimitiveArray<i128>>()
-                    .unwrap();
                 let values = array.values().iter().map(|x| *x as i64);
                 let values = Buffer::from_trusted_len_iter(values);
                 let array = PrimitiveArray::<i64>::from_data(
@@ -275,10 +271,6 @@ pub fn array_to_page(
                 );
                 primitive::array_to_page::<i64, i64>(&array, options, descriptor)
             } else {
-                let array = array
-                    .as_any()
-                    .downcast_ref::<PrimitiveArray<i128>>()
-                    .unwrap();
                 let size = decimal_length_from_precision(precision);
 
                 let mut values = MutableBuffer::<u8>::new(); // todo: this can be estimated
diff --git a/src/io/parquet/write/schema.rs b/src/io/parquet/write/schema.rs
index a25aca99b2..b3169d8785 100644
--- a/src/io/parquet/write/schema.rs
+++ b/src/io/parquet/write/schema.rs
@@ -13,7 +13,7 @@ use crate::{
 use super::super::ARROW_SCHEMA_META_KEY;
 
 pub fn schema_to_metadata_key(schema: &Schema) -> KeyValue {
-    let serialized_schema = schema_to_bytes(&schema, MetadataVersion::V5);
+    let serialized_schema = schema_to_bytes(schema, MetadataVersion::V5);
 
     // manually prepending the length to the schema as arrow uses the legacy IPC format
     // TODO: change after addressing ARROW-9777
diff --git a/src/io/print.rs b/src/io/print.rs
index 9d4a605440..cc4b9ac26a 100644
--- a/src/io/print.rs
+++ b/src/io/print.rs
@@ -44,7 +44,7 @@ fn create_table(results: &[RecordBatch]) -> Result<Table> {
 
     let mut header = Vec::new();
     for field in schema.fields() {
-        header.push(Cell::new(&field.name()));
+        header.push(Cell::new(field.name()));
     }
     table.set_titles(Row::new(header));
 


Test patch:
diff --git a/arrow-pyarrow-integration-testing/tests/test_sql.py b/arrow-pyarrow-integration-testing/tests/test_sql.py
index 2da732b746..e6ec3223d0 100644
--- a/arrow-pyarrow-integration-testing/tests/test_sql.py
+++ b/arrow-pyarrow-integration-testing/tests/test_sql.py
@@ -17,6 +17,7 @@
 # under the License.
 
 import unittest
+import decimal
 
 import pyarrow
 import arrow_pyarrow_integration_testing
@@ -24,12 +25,17 @@ import arrow_pyarrow_integration_testing
 
 class TestCase(unittest.TestCase):
     def setUp(self):
-        self.old_allocated_rust = arrow_pyarrow_integration_testing.total_allocated_bytes()
+        self.old_allocated_rust = (
+            arrow_pyarrow_integration_testing.total_allocated_bytes()
+        )
         self.old_allocated_cpp = pyarrow.total_allocated_bytes()
 
     def tearDown(self):
         # No leak of Rust
-        self.assertEqual(self.old_allocated_rust, arrow_pyarrow_integration_testing.total_allocated_bytes())
+        self.assertEqual(
+            self.old_allocated_rust,
+            arrow_pyarrow_integration_testing.total_allocated_bytes(),
+        )
 
         # No leak of C++ memory
         self.assertEqual(self.old_allocated_cpp, pyarrow.total_allocated_bytes())
@@ -46,6 +52,7 @@ class TestCase(unittest.TestCase):
         """
         Rust -> Python -> Rust
         """
+
         def double(array):
             array = array.to_pylist()
             return pyarrow.array([x * 2 if x is not None else None for x in array])
@@ -90,6 +97,19 @@ class TestCase(unittest.TestCase):
         c = pyarrow.array(["a", None, "ccc"])
         self.assertEqual(b, c)
 
+    def test_decimal_roundtrip(self):
+        """
+        Python -> Rust -> Python
+        """
+        data = [
+            round(decimal.Decimal(722.82), 2),
+            round(decimal.Decimal(-934.11), 2),
+            None
+        ]
+        a = pyarrow.array(data, pyarrow.decimal128(5, 2))
+        b = arrow_pyarrow_integration_testing.round_trip(a)
+        self.assertEqual(a, b)
+
     def test_string_python(self):
         """
         Python -> Rust -> Python
@@ -102,9 +122,9 @@ class TestCase(unittest.TestCase):
         """
         Python -> Rust -> Python
         """
-        a = pyarrow.array([None, 1, 2], pyarrow.time32('s'))
+        a = pyarrow.array([None, 1, 2], pyarrow.time32("s"))
         b = arrow_pyarrow_integration_testing.concatenate(a)
-        expected = pyarrow.array([None, 1, 2] + [None, 1, 2], pyarrow.time32('s'))
+        expected = pyarrow.array([None, 1, 2] + [None, 1, 2], pyarrow.time32("s"))
         self.assertEqual(b, expected)
 
     def test_list_array(self):
@@ -112,7 +132,9 @@ class TestCase(unittest.TestCase):
         Python -> Rust -> Python
         """
         for _ in range(2):
-            a = pyarrow.array([[], None, [1, 2], [4, 5, 6]], pyarrow.list_(pyarrow.int64()))
+            a = pyarrow.array(
+                [[], None, [1, 2], [4, 5, 6]], pyarrow.list_(pyarrow.int64())
+            )
             b = arrow_pyarrow_integration_testing.round_trip(a)
 
             b.validate(full=True)
@@ -124,18 +146,36 @@ class TestCase(unittest.TestCase):
         Python -> Rust -> Python
         """
         fields = [
-            ('f1', pyarrow.int32()),
-            ('f2', pyarrow.string()),
+            ("f1", pyarrow.int32()),
+            ("f2", pyarrow.string()),
         ]
-        a = pyarrow.array([
-            {"f1": 1, "f2": "a"},
-            None,
-            {"f1": 3, "f2": None},
-            {"f1": None, "f2": "d"},
-            {"f1": None, "f2": None},
-        ], pyarrow.struct(fields))
+        a = pyarrow.array(
+            [
+                {"f1": 1, "f2": "a"},
+                None,
+                {"f1": 3, "f2": None},
+                {"f1": None, "f2": "d"},
+                {"f1": None, "f2": None},
+            ],
+            pyarrow.struct(fields),
+        )
         b = arrow_pyarrow_integration_testing.round_trip(a)
 
         b.validate(full=True)
         assert a.to_pylist() == b.to_pylist()
         assert a.type == b.type
+
+    def test_list_list_array(self):
+        """
+        Python -> Rust -> Python
+        """
+        for _ in range(2):
+            a = pyarrow.array(
+                [[None], None, [[1], [2]], [[4, 5], [6]]],
+                pyarrow.list_(pyarrow.list_(pyarrow.int64())),
+            )
+            b = arrow_pyarrow_integration_testing.round_trip(a)
+
+            b.validate(full=True)
+            assert a.to_pylist() == b.to_pylist()
+            assert a.type == b.type