Repository: apache/arrow-rs
Pull request #7047 (2025-01-30 10:02:15)

Add test of out of order UploadPart
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes #.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

There have been various rumblings about concurrency issues related to multipart upload, this is my unsuccessful attempt to create a reproducer. 

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please call them out.
-->



Patch:
diff --git a/README.md b/README.md
index 7be578278..5254afadc 100644
--- a/README.md
+++ b/README.md
@@ -63,14 +63,11 @@ is described in the [contributing] guide.
 
 Planned Release Schedule
 
-| Approximate Date | Version  | Notes                                      |
-| ---------------- | -------- | ------------------------------------------ |
-| Nov 2024         | `53.3.0` | Minor, NO breaking API changes             |
-| Dec 2024         | `54.0.0` | Major, potentially breaking API changes    |
-| Jan 2025         | `53.4.0` | Minor, NO breaking API changes (`53` line) |
-| Jan 2025         | `54.1.0` | Minor, NO breaking API changes             |
-| Feb 2025         | `54.2.0` | Minor, NO breaking API changes             |
-| Mar 2025         | `55.0.0` | Major, potentially breaking API changes    |
+| Approximate Date | Version  | Notes                                   |
+| ---------------- | -------- | --------------------------------------- |
+| Jan 2025         | `54.1.0` | Minor, NO breaking API changes          |
+| Feb 2025         | `54.2.0` | Minor, NO breaking API changes          |
+| Mar 2025         | `55.0.0` | Major, potentially breaking API changes |
 
 [ticket #5368]: https://github.com/apache/arrow-rs/issues/5368
 [semantic versioning]: https://semver.org/
@@ -87,7 +84,6 @@ Planned Release Schedule
 
 | Approximate Date | Version  | Notes                                   |
 | ---------------- | -------- | --------------------------------------- |
-| Dec 2024         | `0.11.2` | Minor, NO breaking API changes          |
 | Feb 2025         | `0.12.0` | Major, potentially breaking API changes |
 
 ### Guidelines for `panic` vs `Result`
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index 6eae8e246..ce66f8a9f 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -55,7 +55,7 @@ ffi = ["arrow-schema/ffi", "arrow-data/ffi"]
 force_validate = []
 
 [dev-dependencies]
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 criterion = { version = "0.5", default-features = false }
 
 [build-dependencies]
diff --git a/arrow-array/benches/fixed_size_list_array.rs b/arrow-array/benches/fixed_size_list_array.rs
index 5270a4a5d..f9b862f88 100644
--- a/arrow-array/benches/fixed_size_list_array.rs
+++ b/arrow-array/benches/fixed_size_list_array.rs
@@ -18,13 +18,13 @@
 use arrow_array::{Array, FixedSizeListArray, Int32Array};
 use arrow_schema::Field;
 use criterion::*;
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 use std::sync::Arc;
 
 fn gen_fsl(len: usize, value_len: usize) -> FixedSizeListArray {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let values = Arc::new(Int32Array::from(
-        (0..len).map(|_| rng.gen::<i32>()).collect::<Vec<_>>(),
+        (0..len).map(|_| rng.random::<i32>()).collect::<Vec<_>>(),
     ));
     let field = Arc::new(Field::new_list_field(values.data_type().clone(), true));
     FixedSizeListArray::new(field, value_len as i32, values, None)
diff --git a/arrow-array/benches/occupancy.rs b/arrow-array/benches/occupancy.rs
index ed4b94351..fd334b613 100644
--- a/arrow-array/benches/occupancy.rs
+++ b/arrow-array/benches/occupancy.rs
@@ -19,7 +19,7 @@ use arrow_array::types::Int32Type;
 use arrow_array::{DictionaryArray, Int32Array};
 use arrow_buffer::NullBuffer;
 use criterion::*;
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 use std::sync::Arc;
 
 fn gen_dict(
@@ -28,11 +28,11 @@ fn gen_dict(
     occupancy: f64,
     null_percent: f64,
 ) -> DictionaryArray<Int32Type> {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let values = Int32Array::from(vec![0; values_len]);
     let max_key = (values_len as f64 * occupancy) as i32;
-    let keys = (0..len).map(|_| rng.gen_range(0..max_key)).collect();
-    let nulls = (0..len).map(|_| !rng.gen_bool(null_percent)).collect();
+    let keys = (0..len).map(|_| rng.random_range(0..max_key)).collect();
+    let nulls = (0..len).map(|_| !rng.random_bool(null_percent)).collect();
 
     let keys = Int32Array::new(keys, Some(NullBuffer::new(nulls)));
     DictionaryArray::new(keys, Arc::new(values))
diff --git a/arrow-array/benches/union_array.rs b/arrow-array/benches/union_array.rs
index c5b2ec0f7..753cc8148 100644
--- a/arrow-array/benches/union_array.rs
+++ b/arrow-array/benches/union_array.rs
@@ -24,17 +24,17 @@ use arrow_array::{Array, ArrayRef, Int32Array, UnionArray};
 use arrow_buffer::{NullBuffer, ScalarBuffer};
 use arrow_schema::{DataType, Field, UnionFields};
 use criterion::*;
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 
 fn array_with_nulls() -> ArrayRef {
-    let mut rng = thread_rng();
+    let mut rng = rng();
 
-    let values = ScalarBuffer::from_iter(repeat_with(|| rng.gen()).take(4096));
+    let values = ScalarBuffer::from_iter(repeat_with(|| rng.random()).take(4096));
 
     // nulls with at least one null and one valid
     let nulls: NullBuffer = [true, false]
         .into_iter()
-        .chain(repeat_with(|| rng.gen()))
+        .chain(repeat_with(|| rng.random()))
         .take(4096)
         .collect();
 
@@ -42,9 +42,9 @@ fn array_with_nulls() -> ArrayRef {
 }
 
 fn array_without_nulls() -> ArrayRef {
-    let mut rng = thread_rng();
+    let mut rng = rng();
 
-    let values = ScalarBuffer::from_iter(repeat_with(|| rng.gen()).take(4096));
+    let values = ScalarBuffer::from_iter(repeat_with(|| rng.random()).take(4096));
 
     Arc::new(Int32Array::new(values.clone(), None))
 }
diff --git a/arrow-array/src/array/boolean_array.rs b/arrow-array/src/array/boolean_array.rs
index 9c2d4af8c..fcebf5a0f 100644
--- a/arrow-array/src/array/boolean_array.rs
+++ b/arrow-array/src/array/boolean_array.rs
@@ -479,7 +479,7 @@ impl From<BooleanBuffer> for BooleanArray {
 mod tests {
     use super::*;
     use arrow_buffer::Buffer;
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
 
     #[test]
     fn test_boolean_fmt_debug() {
@@ -667,11 +667,11 @@ mod tests {
     #[test]
     #[cfg_attr(miri, ignore)] // Takes too long
     fn test_true_false_count() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
         for _ in 0..10 {
             // No nulls
-            let d: Vec<_> = (0..2000).map(|_| rng.gen_bool(0.5)).collect();
+            let d: Vec<_> = (0..2000).map(|_| rng.random_bool(0.5)).collect();
             let b = BooleanArray::from(d.clone());
 
             let expected_true = d.iter().filter(|x| **x).count();
@@ -680,7 +680,7 @@ mod tests {
 
             // With nulls
             let d: Vec<_> = (0..2000)
-                .map(|_| rng.gen_bool(0.5).then(|| rng.gen_bool(0.5)))
+                .map(|_| rng.random_bool(0.5).then(|| rng.random_bool(0.5)))
                 .collect();
             let b = BooleanArray::from(d.clone());
 
diff --git a/arrow-array/src/array/list_array.rs b/arrow-array/src/array/list_array.rs
index b53bbe7ab..009a7b7a5 100644
--- a/arrow-array/src/array/list_array.rs
+++ b/arrow-array/src/array/list_array.rs
@@ -120,8 +120,41 @@ impl OffsetSizeTrait for i64 {
 ///                 (offsets[i],            │   ListArray               (Array)
 ///                offsets[i+1])                                    └ ─ ─ ─ ─ ─ ─ ┘    │
 ///                                         └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+/// ```
+///
+/// # Slicing
+///
+/// Slicing a `ListArray` creates a new `ListArray` without copying any data,
+/// but this means the [`Self::values`] and [`Self::offsets`] may have "unused" data
 ///
+/// For example, calling `slice(1, 3)` on the `ListArray` in the above example
+/// would result in the following. Note
 ///
+/// 1. `Values` array is unchanged
+/// 2. `Offsets` do not start at `0`, nor cover all values in the Values array.
+///
+/// ```text
+///                                 ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+///                                                         ┌ ─ ─ ─ ─ ─ ─ ┐    │  ╔═══╗
+///                                 │                         ╔═══╗ ╔═══╗         ║   ║  Not used
+///                                                         │ ║ 1 ║ ║ A ║ │ 0  │  ╚═══╝
+///  ┌─────────────┐  ┌───────┐     │     ┌───┐   ┌───┐       ╠═══╣ ╠═══╣
+///  │ [] (empty)  │  │ (3,3) │           │ 1 │   │ 3 │     │ ║ 1 ║ ║ B ║ │ 1  │
+///  ├─────────────┤  ├───────┤     │     ├───┤   ├───┤       ╠═══╣ ╠═══╣
+///  │    NULL     │  │ (3,4) │           │ 0 │   │ 3 │     │ ║ 1 ║ ║ C ║ │ 2  │
+///  ├─────────────┤  ├───────┤     │     ├───┤   ├───┤       ╠───╣ ╠───╣
+///  │     [D]     │  │ (4,5) │           │ 1 │   │ 4 │     │ │ 0 │ │ ? │ │ 3  │
+///  └─────────────┘  └───────┘     │     └───┘   ├───┤       ├───┤ ├───┤
+///                                               │ 5 │     │ │ 1 │ │ D │ │ 4  │
+///                                 │             └───┘       ├───┤ ├───┤
+///                                                         │ │ 0 │ │ ? │ │ 5  │
+///                                 │  Validity               ╠═══╣ ╠═══╣
+///     Logical       Logical          (nulls)   Offsets    │ ║ 1 ║ ║ F ║ │ 6  │
+///      Values       Offsets       │                         ╚═══╝ ╚═══╝
+///                                                         │    Values   │    │
+///                 (offsets[i],    │   ListArray               (Array)
+///                offsets[i+1])                            └ ─ ─ ─ ─ ─ ─ ┘    │
+///                                 └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
 /// ```
 ///
 /// [`StringArray`]: crate::array::StringArray
@@ -263,13 +296,22 @@ impl<OffsetSize: OffsetSizeTrait> GenericListArray<OffsetSize> {
     /// Returns a reference to the offsets of this list
     ///
     /// Unlike [`Self::value_offsets`] this returns the [`OffsetBuffer`]
-    /// allowing for zero-copy cloning
+    /// allowing for zero-copy cloning.
+    ///
+    /// Notes: The `offsets` may not start at 0 and may not cover all values in
+    /// [`Self::values`]. This can happen when the list array was sliced via
+    /// [`Self::slice`]. See documentation for [`Self`] for more details.
     #[inline]
     pub fn offsets(&self) -> &OffsetBuffer<OffsetSize> {
         &self.value_offsets
     }
 
     /// Returns a reference to the values of this list
+    ///
+    /// Note: The list array may not refer to all values in the `values` array.
+    /// For example if the list array was sliced via [`Self::slice`] values will
+    /// still contain values both before and after the slice. See documentation
+    /// for [`Self`] for more details.
     #[inline]
     pub fn values(&self) -> &ArrayRef {
         &self.values
@@ -296,7 +338,9 @@ impl<OffsetSize: OffsetSizeTrait> GenericListArray<OffsetSize> {
         self.values.slice(start, end - start)
     }
 
-    /// Returns the offset values in the offsets buffer
+    /// Returns the offset values in the offsets buffer.
+    ///
+    /// See [`Self::offsets`] for more details.
     #[inline]
     pub fn value_offsets(&self) -> &[OffsetSize] {
         &self.value_offsets
@@ -325,6 +369,10 @@ impl<OffsetSize: OffsetSizeTrait> GenericListArray<OffsetSize> {
     }
 
     /// Returns a zero-copy slice of this array with the indicated offset and length.
+    ///
+    /// Notes: this method does *NOT* slice the underlying values array or modify
+    /// the values in the offsets buffer. See [`Self::values`] and
+    /// [`Self::offsets`] for more information.
     pub fn slice(&self, offset: usize, length: usize) -> Self {
         Self {
             data_type: self.data_type.clone(),
@@ -556,12 +604,12 @@ impl<OffsetSize: OffsetSizeTrait> std::fmt::Debug for GenericListArray<OffsetSiz
 
 /// A [`GenericListArray`] of variable size lists, storing offsets as `i32`.
 ///
-// See [`ListBuilder`](crate::builder::ListBuilder) for how to construct a [`ListArray`]
+/// See [`ListBuilder`](crate::builder::ListBuilder) for how to construct a [`ListArray`]
 pub type ListArray = GenericListArray<i32>;
 
 /// A [`GenericListArray`] of variable size lists, storing offsets as `i64`.
 ///
-// See [`LargeListBuilder`](crate::builder::LargeListBuilder) for how to construct a [`LargeListArray`]
+/// See [`LargeListBuilder`](crate::builder::LargeListBuilder) for how to construct a [`LargeListArray`]
 pub type LargeListArray = GenericListArray<i64>;
 
 #[cfg(test)]
diff --git a/arrow-array/src/array/list_view_array.rs b/arrow-array/src/array/list_view_array.rs
index 195ac7e11..6118607bc 100644
--- a/arrow-array/src/array/list_view_array.rs
+++ b/arrow-array/src/array/list_view_array.rs
@@ -895,8 +895,8 @@ mod tests {
                 .build()
                 .unwrap(),
         );
-        assert_eq!(string.value_offsets(), &[]);
-        assert_eq!(string.value_sizes(), &[]);
+        assert_eq!(string.value_offsets(), &[] as &[i32; 0]);
+        assert_eq!(string.value_sizes(), &[] as &[i32; 0]);
 
         let string = LargeListViewArray::from(
             ArrayData::builder(DataType::LargeListView(f))
@@ -906,8 +906,8 @@ mod tests {
                 .unwrap(),
         );
         assert_eq!(string.len(), 0);
-        assert_eq!(string.value_offsets(), &[]);
-        assert_eq!(string.value_sizes(), &[]);
+        assert_eq!(string.value_offsets(), &[] as &[i64; 0]);
+        assert_eq!(string.value_sizes(), &[] as &[i64; 0]);
     }
 
     #[test]
diff --git a/arrow-array/src/array/run_array.rs b/arrow-array/src/array/run_array.rs
index b340bf9a9..6d909cbeb 100644
--- a/arrow-array/src/array/run_array.rs
+++ b/arrow-array/src/array/run_array.rs
@@ -662,8 +662,8 @@ where
 
 #[cfg(test)]
 mod tests {
+    use rand::rng;
     use rand::seq::SliceRandom;
-    use rand::thread_rng;
     use rand::Rng;
 
     use super::*;
@@ -691,7 +691,7 @@ mod tests {
         ];
         let mut result: Vec<Option<i32>> = Vec::with_capacity(size);
         let mut ix = 0;
-        let mut rng = thread_rng();
+        let mut rng = rng();
         // run length can go up to 8. Cap the max run length for smaller arrays to size / 2.
         let max_run_length = 8_usize.min(1_usize.max(size / 2));
         while result.len() < size {
@@ -700,7 +700,7 @@ mod tests {
                 seed.shuffle(&mut rng);
             }
             // repeat the items between 1 and 8 times. Cap the length for smaller sized arrays
-            let num = max_run_length.min(rand::thread_rng().gen_range(1..=max_run_length));
+            let num = max_run_length.min(rand::rng().random_range(1..=max_run_length));
             for _ in 0..num {
                 result.push(seed[ix]);
             }
@@ -1000,7 +1000,7 @@ mod tests {
             let mut logical_indices: Vec<u32> = (0_u32..(logical_len as u32)).collect();
             // add same indices once more
             logical_indices.append(&mut logical_indices.clone());
-            let mut rng = thread_rng();
+            let mut rng = rng();
             logical_indices.shuffle(&mut rng);
 
             let physical_indices = run_array.get_physical_indices(&logical_indices).unwrap();
@@ -1036,7 +1036,7 @@ mod tests {
             let mut logical_indices: Vec<u32> = (0_u32..(slice_len as u32)).collect();
             // add same indices once more
             logical_indices.append(&mut logical_indices.clone());
-            let mut rng = thread_rng();
+            let mut rng = rng();
             logical_indices.shuffle(&mut rng);
 
             // test for offset = 0 and slice length = slice_len
diff --git a/arrow-array/src/run_iterator.rs b/arrow-array/src/run_iterator.rs
index 2922bf04d..f758f749f 100644
--- a/arrow-array/src/run_iterator.rs
+++ b/arrow-array/src/run_iterator.rs
@@ -172,7 +172,7 @@ where
 
 #[cfg(test)]
 mod tests {
-    use rand::{seq::SliceRandom, thread_rng, Rng};
+    use rand::{rng, seq::SliceRandom, Rng};
 
     use crate::{
         array::{Int32Array, StringArray},
@@ -200,7 +200,7 @@ mod tests {
         ];
         let mut result: Vec<Option<i32>> = Vec::with_capacity(size);
         let mut ix = 0;
-        let mut rng = thread_rng();
+        let mut rng = rng();
         // run length can go up to 8. Cap the max run length for smaller arrays to size / 2.
         let max_run_length = 8_usize.min(1_usize.max(size / 2));
         while result.len() < size {
@@ -209,7 +209,7 @@ mod tests {
                 seed.shuffle(&mut rng);
             }
             // repeat the items between 1 and 8 times. Cap the length for smaller sized arrays
-            let num = max_run_length.min(rand::thread_rng().gen_range(1..=max_run_length));
+            let num = max_run_length.min(rand::rng().random_range(1..=max_run_length));
             for _ in 0..num {
                 result.push(seed[ix]);
             }
diff --git a/arrow-avro/Cargo.toml b/arrow-avro/Cargo.toml
index c103c2ecc..b1391559f 100644
--- a/arrow-avro/Cargo.toml
+++ b/arrow-avro/Cargo.toml
@@ -51,5 +51,4 @@ crc = { version = "3.0", optional = true }
 
 
 [dev-dependencies]
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
diff --git a/arrow-buffer/Cargo.toml b/arrow-buffer/Cargo.toml
index 68bfe8ddf..69029759d 100644
--- a/arrow-buffer/Cargo.toml
+++ b/arrow-buffer/Cargo.toml
@@ -40,7 +40,7 @@ half = { version = "2.1", default-features = false }
 
 [dev-dependencies]
 criterion = { version = "0.5", default-features = false }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 
 [build-dependencies]
 
diff --git a/arrow-buffer/benches/i256.rs b/arrow-buffer/benches/i256.rs
index ebb45e793..f94ca6951 100644
--- a/arrow-buffer/benches/i256.rs
+++ b/arrow-buffer/benches/i256.rs
@@ -47,8 +47,8 @@ fn criterion_benchmark(c: &mut Criterion) {
 
     let numerators: Vec<_> = (0..SIZE)
         .map(|_| {
-            let high = rng.gen_range(1000..i128::MAX);
-            let low = rng.gen();
+            let high = rng.random_range(1000..i128::MAX);
+            let low = rng.random();
             i256::from_parts(low, high)
         })
         .collect();
@@ -56,7 +56,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     let divisors: Vec<_> = numerators
         .iter()
         .map(|n| {
-            let quotient = rng.gen_range(1..100_i32);
+            let quotient = rng.random_range(1..100_i32);
             n.wrapping_div(i256::from(quotient))
         })
         .collect();
@@ -70,7 +70,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 
     let divisors: Vec<_> = (0..SIZE)
-        .map(|_| i256::from(rng.gen_range(1..100_i32)))
+        .map(|_| i256::from(rng.random_range(1..100_i32)))
         .collect();
 
     c.bench_function("i256_div_rem small divisor", |b| {
diff --git a/arrow-buffer/benches/offset.rs b/arrow-buffer/benches/offset.rs
index 1aea5024f..12c00a60c 100644
--- a/arrow-buffer/benches/offset.rs
+++ b/arrow-buffer/benches/offset.rs
@@ -24,7 +24,7 @@ const SIZE: usize = 1024;
 
 fn criterion_benchmark(c: &mut Criterion) {
     let mut rng = StdRng::seed_from_u64(42);
-    let lengths: Vec<usize> = black_box((0..SIZE).map(|_| rng.gen_range(0..40)).collect());
+    let lengths: Vec<usize> = black_box((0..SIZE).map(|_| rng.random_range(0..40)).collect());
 
     c.bench_function("OffsetBuffer::from_lengths", |b| {
         b.iter(|| OffsetBuffer::<i32>::from_lengths(lengths.iter().copied()));
diff --git a/arrow-buffer/src/bigint/mod.rs b/arrow-buffer/src/bigint/mod.rs
index f5fab75dc..b4f2e2bdd 100644
--- a/arrow-buffer/src/bigint/mod.rs
+++ b/arrow-buffer/src/bigint/mod.rs
@@ -840,7 +840,7 @@ impl ToPrimitive for i256 {
 mod tests {
     use super::*;
     use num::Signed;
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
 
     #[test]
     fn test_signed_cmp() {
@@ -1091,16 +1091,16 @@ mod tests {
     #[test]
     #[cfg_attr(miri, ignore)]
     fn test_i256_fuzz() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
         for _ in 0..1000 {
             let mut l = [0_u8; 32];
-            let len = rng.gen_range(0..32);
-            l.iter_mut().take(len).for_each(|x| *x = rng.gen());
+            let len = rng.random_range(0..32);
+            l.iter_mut().take(len).for_each(|x| *x = rng.random());
 
             let mut r = [0_u8; 32];
-            let len = rng.gen_range(0..32);
-            r.iter_mut().take(len).for_each(|x| *x = rng.gen());
+            let len = rng.random_range(0..32);
+            r.iter_mut().take(len).for_each(|x| *x = rng.random());
 
             test_ops(i256::from_le_bytes(l), i256::from_le_bytes(r))
         }
diff --git a/arrow-buffer/src/buffer/scalar.rs b/arrow-buffer/src/buffer/scalar.rs
index ab6c87168..94dd7bcfc 100644
--- a/arrow-buffer/src/buffer/scalar.rs
+++ b/arrow-buffer/src/buffer/scalar.rs
@@ -182,6 +182,7 @@ impl<T: ArrowNativeType> From<BufferBuilder<T>> for ScalarBuffer<T> {
 }
 
 impl<T: ArrowNativeType> FromIterator<T> for ScalarBuffer<T> {
+    #[inline]
     fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
         iter.into_iter().collect::<Vec<_>>().into()
     }
diff --git a/arrow-buffer/src/builder/boolean.rs b/arrow-buffer/src/builder/boolean.rs
index 0424e128d..da8fb0643 100644
--- a/arrow-buffer/src/builder/boolean.rs
+++ b/arrow-buffer/src/builder/boolean.rs
@@ -399,7 +399,7 @@ mod tests {
 
         let mut buffer = BooleanBufferBuilder::new(12);
         let mut all_bools = vec![];
-        let mut rng = rand::thread_rng();
+        let mut rng = rand::rng();
 
         let src_len = 32;
         let (src, compacted_src) = {
diff --git a/arrow-buffer/src/util/bit_chunk_iterator.rs b/arrow-buffer/src/util/bit_chunk_iterator.rs
index 54995314c..0562c7ad5 100644
--- a/arrow-buffer/src/util/bit_chunk_iterator.rs
+++ b/arrow-buffer/src/util/bit_chunk_iterator.rs
@@ -371,7 +371,10 @@ impl ExactSizeIterator for BitChunkIterator<'_> {
 
 #[cfg(test)]
 mod tests {
+    use rand::distr::uniform::UniformSampler;
+    use rand::distr::uniform::UniformUsize;
     use rand::prelude::*;
+    use rand::rng;
 
     use crate::buffer::Buffer;
     use crate::util::bit_chunk_iterator::UnalignedBitChunk;
@@ -624,21 +627,25 @@ mod tests {
     #[test]
     #[cfg_attr(miri, ignore)]
     fn fuzz_unaligned_bit_chunk_iterator() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
+        let uusize = UniformUsize::new(usize::MIN, usize::MAX).unwrap();
         for _ in 0..100 {
-            let mask_len = rng.gen_range(0..1024);
-            let bools: Vec<_> = std::iter::from_fn(|| Some(rng.gen()))
+            let mask_len = rng.random_range(0..1024);
+            let bools: Vec<_> = std::iter::from_fn(|| Some(rng.random()))
                 .take(mask_len)
                 .collect();
 
             let buffer = Buffer::from_iter(bools.iter().cloned());
 
             let max_offset = 64.min(mask_len);
-            let offset = rng.gen::<usize>().checked_rem(max_offset).unwrap_or(0);
+            let offset = uusize.sample(&mut rng).checked_rem(max_offset).unwrap_or(0);
 
             let max_truncate = 128.min(mask_len - offset);
-            let truncate = rng.gen::<usize>().checked_rem(max_truncate).unwrap_or(0);
+            let truncate = uusize
+                .sample(&mut rng)
+                .checked_rem(max_truncate)
+                .unwrap_or(0);
 
             let unaligned =
                 UnalignedBitChunk::new(buffer.as_slice(), offset, mask_len - offset - truncate);
diff --git a/arrow-buffer/src/util/bit_mask.rs b/arrow-buffer/src/util/bit_mask.rs
index 97be7e006..0d694d13e 100644
--- a/arrow-buffer/src/util/bit_mask.rs
+++ b/arrow-buffer/src/util/bit_mask.rs
@@ -164,7 +164,7 @@ mod tests {
     use super::*;
     use crate::bit_util::{get_bit, set_bit, unset_bit};
     use rand::prelude::StdRng;
-    use rand::{Fill, Rng, SeedableRng};
+    use rand::{Rng, SeedableRng, TryRngCore};
     use std::fmt::Display;
 
     #[test]
@@ -322,20 +322,20 @@ mod tests {
             // -------------------+-----------------+-------
 
             // length of data to copy
-            let len = rng.gen_range(0..=200);
+            let len = rng.random_range(0..=200);
 
             // randomly pick where we will write to
-            let offset_write_bits = rng.gen_range(0..=200);
+            let offset_write_bits = rng.random_range(0..=200);
             let offset_write_bytes = if offset_write_bits % 8 == 0 {
                 offset_write_bits / 8
             } else {
                 (offset_write_bits / 8) + 1
             };
-            let extra_write_data_bytes = rng.gen_range(0..=5); // ensure 0 shows up often
+            let extra_write_data_bytes = rng.random_range(0..=5); // ensure 0 shows up often
 
             // randomly decide where we will read from
-            let extra_read_data_bytes = rng.gen_range(0..=5); // make sure 0 shows up often
-            let offset_read_bits = rng.gen_range(0..=200);
+            let extra_read_data_bytes = rng.random_range(0..=5); // make sure 0 shows up often
+            let offset_read_bits = rng.random_range(0..=200);
             let offset_read_bytes = if offset_read_bits % 8 != 0 {
                 (offset_read_bits / 8) + 1
             } else {
@@ -356,7 +356,7 @@ mod tests {
             self.data
                 .resize(offset_read_bytes + len + extra_read_data_bytes, 0);
             // fill source data with random bytes
-            self.data.try_fill(rng).unwrap();
+            rng.try_fill_bytes(self.data.as_mut_slice()).unwrap();
             self.offset_read = offset_read_bits;
 
             self.len = len;
diff --git a/arrow-buffer/src/util/bit_util.rs b/arrow-buffer/src/util/bit_util.rs
index ed5d363d6..f39cb69c3 100644
--- a/arrow-buffer/src/util/bit_util.rs
+++ b/arrow-buffer/src/util/bit_util.rs
@@ -153,7 +153,7 @@ mod tests {
         let mut expected = vec![];
         let mut rng = seedable_rng();
         for i in 0..8 * NUM_BYTE {
-            let b = rng.gen_bool(0.5);
+            let b = rng.random_bool(0.5);
             expected.push(b);
             if b {
                 set_bit(&mut buf[..], i)
@@ -197,7 +197,7 @@ mod tests {
         let mut expected = vec![];
         let mut rng = seedable_rng();
         for i in 0..8 * NUM_BYTE {
-            let b = rng.gen_bool(0.5);
+            let b = rng.random_bool(0.5);
             expected.push(b);
             if b {
                 unsafe {
@@ -221,7 +221,7 @@ mod tests {
         let mut expected = vec![];
         let mut rng = seedable_rng();
         for i in 0..8 * NUM_BYTE {
-            let b = rng.gen_bool(0.5);
+            let b = rng.random_bool(0.5);
             expected.push(b);
             if !b {
                 unsafe {
@@ -247,7 +247,7 @@ mod tests {
         let mut v = HashSet::new();
         let mut rng = seedable_rng();
         for _ in 0..NUM_SETS {
-            let offset = rng.gen_range(0..8 * NUM_BYTES);
+            let offset = rng.random_range(0..8 * NUM_BYTES);
             v.insert(offset);
             set_bit(&mut buffer[..], offset);
         }
diff --git a/arrow-cast/Cargo.toml b/arrow-cast/Cargo.toml
index 4046f5226..22e446ce3 100644
--- a/arrow-cast/Cargo.toml
+++ b/arrow-cast/Cargo.toml
@@ -58,7 +58,7 @@ ryu = "1.0.16"
 [dev-dependencies]
 criterion = { version = "0.5", default-features = false }
 half = { version = "2.1", default-features = false }
-rand = "0.8"
+rand = "0.9"
 
 [build-dependencies]
 
diff --git a/arrow-cast/src/base64.rs b/arrow-cast/src/base64.rs
index 534b21878..e7bb84ebe 100644
--- a/arrow-cast/src/base64.rs
+++ b/arrow-cast/src/base64.rs
@@ -90,7 +90,7 @@ pub fn b64_decode<E: Engine, O: OffsetSizeTrait>(
 mod tests {
     use super::*;
     use arrow_array::BinaryArray;
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
 
     fn test_engine<E: Engine>(e: &E, a: &BinaryArray) {
         let encoded = b64_encode(e, a);
@@ -105,12 +105,12 @@ mod tests {
 
     #[test]
     fn test_b64() {
-        let mut rng = thread_rng();
-        let len = rng.gen_range(1024..1050);
+        let mut rng = rng();
+        let len = rng.random_range(1024..1050);
         let data: BinaryArray = (0..len)
             .map(|_| {
-                let len = rng.gen_range(0..16);
-                Some((0..len).map(|_| rng.gen()).collect::<Vec<u8>>())
+                let len = rng.random_range(0..16);
+                Some((0..len).map(|_| rng.random()).collect::<Vec<u8>>())
             })
             .collect();
 
diff --git a/arrow-json/Cargo.toml b/arrow-json/Cargo.toml
index 564cb9433..3a2eca218 100644
--- a/arrow-json/Cargo.toml
+++ b/arrow-json/Cargo.toml
@@ -54,9 +54,8 @@ futures = "0.3"
 tokio = { version = "1.27", default-features = false, features = ["io-util"] }
 bytes = "1.4"
 criterion = { version = "0.5", default-features = false }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 
 [[bench]]
 name = "serde"
 harness = false
-
diff --git a/arrow-json/benches/serde.rs b/arrow-json/benches/serde.rs
index 7636b9c9d..7baaac458 100644
--- a/arrow-json/benches/serde.rs
+++ b/arrow-json/benches/serde.rs
@@ -18,7 +18,7 @@
 use arrow_json::ReaderBuilder;
 use arrow_schema::{DataType, Field, Schema};
 use criterion::*;
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 use serde::Serialize;
 use std::sync::Arc;
 
@@ -35,26 +35,28 @@ fn do_bench<R: Serialize>(c: &mut Criterion, name: &str, rows: &[R], schema: &Sc
 }
 
 fn criterion_benchmark(c: &mut Criterion) {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let schema = Schema::new(vec![Field::new("i32", DataType::Int32, false)]);
-    let v: Vec<i32> = (0..2048).map(|_| rng.gen_range(0..10000)).collect();
+    let v: Vec<i32> = (0..2048).map(|_| rng.random_range(0..10000)).collect();
 
     do_bench(c, "small_i32", &v, &schema);
-    let v: Vec<i32> = (0..2048).map(|_| rng.gen()).collect();
+    let v: Vec<i32> = (0..2048).map(|_| rng.random()).collect();
     do_bench(c, "large_i32", &v, &schema);
 
     let schema = Schema::new(vec![Field::new("i64", DataType::Int64, false)]);
-    let v: Vec<i64> = (0..2048).map(|_| rng.gen_range(0..10000)).collect();
+    let v: Vec<i64> = (0..2048).map(|_| rng.random_range(0..10000)).collect();
     do_bench(c, "small_i64", &v, &schema);
-    let v: Vec<i64> = (0..2048).map(|_| rng.gen_range(0..i32::MAX as _)).collect();
+    let v: Vec<i64> = (0..2048)
+        .map(|_| rng.random_range(0..i32::MAX as _))
+        .collect();
     do_bench(c, "medium_i64", &v, &schema);
-    let v: Vec<i64> = (0..2048).map(|_| rng.gen()).collect();
+    let v: Vec<i64> = (0..2048).map(|_| rng.random()).collect();
     do_bench(c, "large_i64", &v, &schema);
 
     let schema = Schema::new(vec![Field::new("f32", DataType::Float32, false)]);
-    let v: Vec<f32> = (0..2048).map(|_| rng.gen_range(0.0..10000.)).collect();
+    let v: Vec<f32> = (0..2048).map(|_| rng.random_range(0.0..10000.)).collect();
     do_bench(c, "small_f32", &v, &schema);
-    let v: Vec<f32> = (0..2048).map(|_| rng.gen_range(0.0..f32::MAX)).collect();
+    let v: Vec<f32> = (0..2048).map(|_| rng.random_range(0.0..f32::MAX)).collect();
     do_bench(c, "large_f32", &v, &schema);
 }
 
diff --git a/arrow-ord/Cargo.toml b/arrow-ord/Cargo.toml
index 8d74d2f97..560b19b3f 100644
--- a/arrow-ord/Cargo.toml
+++ b/arrow-ord/Cargo.toml
@@ -42,4 +42,4 @@ arrow-select = { workspace = true }
 
 [dev-dependencies]
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng"] }
diff --git a/arrow-ord/src/sort.rs b/arrow-ord/src/sort.rs
index fa5e2b8b2..92ee93e1b 100644
--- a/arrow-ord/src/sort.rs
+++ b/arrow-ord/src/sort.rs
@@ -4281,7 +4281,7 @@ mod tests {
     fn test_partial_rand_sort() {
         let size = 1000u32;
         let mut rng = StdRng::seed_from_u64(42);
-        let mut before: Vec<u32> = (0..size).map(|_| rng.gen::<u32>()).collect();
+        let mut before: Vec<u32> = (0..size).map(|_| rng.random::<u32>()).collect();
         let mut d = before.clone();
         let last = (rng.next_u32() % size) as usize;
         d.sort_unstable();
diff --git a/arrow-row/Cargo.toml b/arrow-row/Cargo.toml
index 90d99684d..2e97abbbc 100644
--- a/arrow-row/Cargo.toml
+++ b/arrow-row/Cargo.toml
@@ -44,7 +44,6 @@ half = { version = "2.1", default-features = false }
 [dev-dependencies]
 arrow-cast = { workspace = true }
 arrow-ord = { workspace = true }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 
 [features]
-
diff --git a/arrow-row/src/lib.rs b/arrow-row/src/lib.rs
index d0fad1221..0e075d587 100644
--- a/arrow-row/src/lib.rs
+++ b/arrow-row/src/lib.rs
@@ -1433,9 +1433,9 @@ unsafe fn decode_column(
 
 #[cfg(test)]
 mod tests {
-    use rand::distributions::uniform::SampleUniform;
-    use rand::distributions::{Distribution, Standard};
-    use rand::{thread_rng, Rng};
+    use rand::distr::uniform::SampleUniform;
+    use rand::distr::{Distribution, StandardUniform};
+    use rand::{rng, Rng};
 
     use arrow_array::builder::*;
     use arrow_array::types::*;
@@ -2193,11 +2193,11 @@ mod tests {
     fn generate_primitive_array<K>(len: usize, valid_percent: f64) -> PrimitiveArray<K>
     where
         K: ArrowPrimitiveType,
-        Standard: Distribution<K::Native>,
+        StandardUniform: Distribution<K::Native>,
     {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         (0..len)
-            .map(|_| rng.gen_bool(valid_percent).then(|| rng.gen()))
+            .map(|_| rng.random_bool(valid_percent).then(|| rng.random()))
             .collect()
     }
 
@@ -2205,12 +2205,12 @@ mod tests {
         len: usize,
         valid_percent: f64,
     ) -> GenericStringArray<O> {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         (0..len)
             .map(|_| {
-                rng.gen_bool(valid_percent).then(|| {
-                    let len = rng.gen_range(0..100);
-                    let bytes = (0..len).map(|_| rng.gen_range(0..128)).collect();
+                rng.random_bool(valid_percent).then(|| {
+                    let len = rng.random_range(0..100);
+                    let bytes = (0..len).map(|_| rng.random_range(0..128)).collect();
                     String::from_utf8(bytes).unwrap()
                 })
             })
@@ -2218,12 +2218,12 @@ mod tests {
     }
 
     fn generate_string_view(len: usize, valid_percent: f64) -> StringViewArray {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         (0..len)
             .map(|_| {
-                rng.gen_bool(valid_percent).then(|| {
-                    let len = rng.gen_range(0..100);
-                    let bytes = (0..len).map(|_| rng.gen_range(0..128)).collect();
+                rng.random_bool(valid_percent).then(|| {
+                    let len = rng.random_range(0..100);
+                    let bytes = (0..len).map(|_| rng.random_range(0..128)).collect();
                     String::from_utf8(bytes).unwrap()
                 })
             })
@@ -2231,12 +2231,12 @@ mod tests {
     }
 
     fn generate_byte_view(len: usize, valid_percent: f64) -> BinaryViewArray {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         (0..len)
             .map(|_| {
-                rng.gen_bool(valid_percent).then(|| {
-                    let len = rng.gen_range(0..100);
-                    let bytes: Vec<_> = (0..len).map(|_| rng.gen_range(0..128)).collect();
+                rng.random_bool(valid_percent).then(|| {
+                    let len = rng.random_range(0..100);
+                    let bytes: Vec<_> = (0..len).map(|_| rng.random_range(0..128)).collect();
                     bytes
                 })
             })
@@ -2252,13 +2252,13 @@ mod tests {
         K: ArrowDictionaryKeyType,
         K::Native: SampleUniform,
     {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         let min_key = K::Native::from_usize(0).unwrap();
         let max_key = K::Native::from_usize(values.len()).unwrap();
         let keys: PrimitiveArray<K> = (0..len)
             .map(|_| {
-                rng.gen_bool(valid_percent)
-                    .then(|| rng.gen_range(min_key..max_key))
+                rng.random_bool(valid_percent)
+                    .then(|| rng.random_range(min_key..max_key))
             })
             .collect();
 
@@ -2277,15 +2277,15 @@ mod tests {
     }
 
     fn generate_fixed_size_binary(len: usize, valid_percent: f64) -> FixedSizeBinaryArray {
-        let mut rng = thread_rng();
-        let width = rng.gen_range(0..20);
+        let mut rng = rng();
+        let width = rng.random_range(0..20);
         let mut builder = FixedSizeBinaryBuilder::new(width);
 
         let mut b = vec![0; width as usize];
         for _ in 0..len {
-            match rng.gen_bool(valid_percent) {
+            match rng.random_bool(valid_percent) {
                 true => {
-                    b.iter_mut().for_each(|x| *x = rng.gen());
+                    b.iter_mut().for_each(|x| *x = rng.random());
                     builder.append_value(&b).unwrap();
                 }
                 false => builder.append_null(),
@@ -2296,8 +2296,8 @@ mod tests {
     }
 
     fn generate_struct(len: usize, valid_percent: f64) -> StructArray {
-        let mut rng = thread_rng();
-        let nulls = NullBuffer::from_iter((0..len).map(|_| rng.gen_bool(valid_percent)));
+        let mut rng = rng();
+        let nulls = NullBuffer::from_iter((0..len).map(|_| rng.random_bool(valid_percent)));
         let a = generate_primitive_array::<Int32Type>(len, valid_percent);
         let b = generate_strings::<i32>(len, valid_percent);
         let fields = Fields::from(vec![
@@ -2312,18 +2312,18 @@ mod tests {
     where
         F: FnOnce(usize) -> ArrayRef,
     {
-        let mut rng = thread_rng();
-        let offsets = OffsetBuffer::<i32>::from_lengths((0..len).map(|_| rng.gen_range(0..10)));
+        let mut rng = rng();
+        let offsets = OffsetBuffer::<i32>::from_lengths((0..len).map(|_| rng.random_range(0..10)));
         let values_len = offsets.last().unwrap().to_usize().unwrap();
         let values = values(values_len);
-        let nulls = NullBuffer::from_iter((0..len).map(|_| rng.gen_bool(valid_percent)));
+        let nulls = NullBuffer::from_iter((0..len).map(|_| rng.random_bool(valid_percent)));
         let field = Arc::new(Field::new_list_field(values.data_type().clone(), true));
         ListArray::new(field, offsets, values, Some(nulls))
     }
 
     fn generate_column(len: usize) -> ArrayRef {
-        let mut rng = thread_rng();
-        match rng.gen_range(0..16) {
+        let mut rng = rng();
+        match rng.random_range(0..16) {
             0 => Arc::new(generate_primitive_array::<Int32Type>(len, 0.8)),
             1 => Arc::new(generate_primitive_array::<UInt32Type>(len, 0.8)),
             2 => Arc::new(generate_primitive_array::<Int64Type>(len, 0.8)),
@@ -2333,14 +2333,14 @@ mod tests {
             6 => Arc::new(generate_strings::<i32>(len, 0.8)),
             7 => Arc::new(generate_dictionary::<Int64Type>(
                 // Cannot test dictionaries containing null values because of #2687
-                Arc::new(generate_strings::<i32>(rng.gen_range(1..len), 1.0)),
+                Arc::new(generate_strings::<i32>(rng.random_range(1..len), 1.0)),
                 len,
                 0.8,
             )),
             8 => Arc::new(generate_dictionary::<Int64Type>(
                 // Cannot test dictionaries containing null values because of #2687
                 Arc::new(generate_primitive_array::<Int64Type>(
-                    rng.gen_range(1..len),
+                    rng.random_range(1..len),
                     1.0,
                 )),
                 len,
@@ -2390,15 +2390,15 @@ mod tests {
     #[cfg_attr(miri, ignore)]
     fn fuzz_test() {
         for _ in 0..100 {
-            let mut rng = thread_rng();
-            let num_columns = rng.gen_range(1..5);
-            let len = rng.gen_range(5..100);
+            let mut rng = rng();
+            let num_columns = rng.random_range(1..5);
+            let len = rng.random_range(5..100);
             let arrays: Vec<_> = (0..num_columns).map(|_| generate_column(len)).collect();
 
             let options: Vec<_> = (0..num_columns)
                 .map(|_| SortOptions {
-                    descending: rng.gen_bool(0.5),
-                    nulls_first: rng.gen_bool(0.5),
+                    descending: rng.random_bool(0.5),
+                    nulls_first: rng.random_bool(0.5),
                 })
                 .collect();
 
diff --git a/arrow-schema/Cargo.toml b/arrow-schema/Cargo.toml
index d1bcf046b..ffea42db6 100644
--- a/arrow-schema/Cargo.toml
+++ b/arrow-schema/Cargo.toml
@@ -34,21 +34,27 @@ path = "src/lib.rs"
 bench = false
 
 [dependencies]
-serde = { version = "1.0", default-features = false, features = ["derive", "std", "rc"], optional = true }
+serde = { version = "1.0", default-features = false, features = [
+    "derive",
+    "std",
+    "rc",
+], optional = true }
 bitflags = { version = "2.0.0", default-features = false, optional = true }
+serde_json = { version = "1.0", optional = true }
 
 [features]
+canonical_extension_types = ["dep:serde", "dep:serde_json"]
 # Enable ffi support
 ffi = ["bitflags"]
+serde = ["dep:serde"]
 
 [package.metadata.docs.rs]
 features = ["ffi"]
 
 [dev-dependencies]
-serde_json = "1.0"
 bincode = { version = "1.3.3", default-features = false }
 criterion = { version = "0.5", default-features = false }
 
 [[bench]]
 name = "ffi"
-harness = false
\ No newline at end of file
+harness = false
diff --git a/arrow-schema/src/extension/canonical/bool8.rs b/arrow-schema/src/extension/canonical/bool8.rs
new file mode 100644
index 000000000..3f6c50cb3
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/bool8.rs
@@ -0,0 +1,142 @@
+// 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.
+
+//! 8-bit Boolean
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#bit-boolean>
+
+use crate::{extension::ExtensionType, ArrowError, DataType};
+
+/// The extension type for `8-bit Boolean`.
+///
+/// Extension name: `arrow.bool8`.
+///
+/// The storage type of the extension is `Int8` where:
+/// - false is denoted by the value 0.
+/// - true can be specified using any non-zero value. Preferably 1.
+///
+/// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#bit-boolean>
+#[derive(Debug, Default, Clone, Copy, PartialEq)]
+pub struct Bool8;
+
+impl ExtensionType for Bool8 {
+    const NAME: &'static str = "arrow.bool8";
+
+    type Metadata = &'static str;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &""
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        Some(String::default())
+    }
+
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+        if metadata.map_or(false, str::is_empty) {
+            Ok("")
+        } else {
+            Err(ArrowError::InvalidArgumentError(
+                "Bool8 extension type expects an empty string as metadata".to_owned(),
+            ))
+        }
+    }
+
+    fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError> {
+        match data_type {
+            DataType::Int8 => Ok(()),
+            data_type => Err(ArrowError::InvalidArgumentError(format!(
+                "Bool8 data type mismatch, expected Int8, found {data_type}"
+            ))),
+        }
+    }
+
+    fn try_new(data_type: &DataType, _metadata: Self::Metadata) -> Result<Self, ArrowError> {
+        Self.supports_data_type(data_type).map(|_| Self)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "canonical_extension_types")]
+    use crate::extension::CanonicalExtensionType;
+    use crate::{
+        extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+        Field,
+    };
+
+    use super::*;
+
+    #[test]
+    fn valid() -> Result<(), ArrowError> {
+        let mut field = Field::new("", DataType::Int8, false);
+        field.try_with_extension_type(Bool8)?;
+        field.try_extension_type::<Bool8>()?;
+        #[cfg(feature = "canonical_extension_types")]
+        assert_eq!(
+            field.try_canonical_extension_type()?,
+            CanonicalExtensionType::Bool8(Bool8)
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "Field extension type name missing")]
+    fn missing_name() {
+        let field = Field::new("", DataType::Int8, false).with_metadata(
+            [(EXTENSION_TYPE_METADATA_KEY.to_owned(), "".to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        field.extension_type::<Bool8>();
+    }
+
+    #[test]
+    #[should_panic(expected = "expected Int8, found Boolean")]
+    fn invalid_type() {
+        Field::new("", DataType::Boolean, false).with_extension_type(Bool8);
+    }
+
+    #[test]
+    #[should_panic(expected = "Bool8 extension type expects an empty string as metadata")]
+    fn missing_metadata() {
+        let field = Field::new("", DataType::Int8, false).with_metadata(
+            [(EXTENSION_TYPE_NAME_KEY.to_owned(), Bool8::NAME.to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        field.extension_type::<Bool8>();
+    }
+
+    #[test]
+    #[should_panic(expected = "Bool8 extension type expects an empty string as metadata")]
+    fn invalid_metadata() {
+        let field = Field::new("", DataType::Int8, false).with_metadata(
+            [
+                (EXTENSION_TYPE_NAME_KEY.to_owned(), Bool8::NAME.to_owned()),
+                (
+                    EXTENSION_TYPE_METADATA_KEY.to_owned(),
+                    "non-empty".to_owned(),
+                ),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<Bool8>();
+    }
+}
diff --git a/arrow-schema/src/extension/canonical/fixed_shape_tensor.rs b/arrow-schema/src/extension/canonical/fixed_shape_tensor.rs
new file mode 100644
index 000000000..6fe94fba7
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/fixed_shape_tensor.rs
@@ -0,0 +1,443 @@
+// 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.
+
+//! FixedShapeTensor
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#fixed-shape-tensor>
+
+use serde::{Deserialize, Serialize};
+
+use crate::{extension::ExtensionType, ArrowError, DataType};
+
+/// The extension type for fixed shape tensor.
+///
+/// Extension name: `arrow.fixed_shape_tensor`.
+///
+/// The storage type of the extension: `FixedSizeList` where:
+/// - `value_type` is the data type of individual tensor elements.
+/// - `list_size` is the product of all the elements in tensor shape.
+///
+/// Extension type parameters:
+/// - `value_type`: the Arrow data type of individual tensor elements.
+/// - `shape`: the physical shape of the contained tensors as an array.
+///
+/// Optional parameters describing the logical layout:
+/// - `dim_names`: explicit names to tensor dimensions as an array. The
+///   length of it should be equal to the shape length and equal to the
+///   number of dimensions.
+///   `dim_names` can be used if the dimensions have
+///   well-known names and they map to the physical layout (row-major).
+/// - `permutation`: indices of the desired ordering of the original
+///   dimensions, defined as an array.
+///   The indices contain a permutation of the values `[0, 1, .., N-1]`
+///   where `N` is the number of dimensions. The permutation indicates
+///   which dimension of the logical layout corresponds to which dimension
+///   of the physical tensor (the i-th dimension of the logical view
+///   corresponds to the dimension with number `permutations[i]` of the
+///   physical tensor).
+///   Permutation can be useful in case the logical order of the tensor is
+///   a permutation of the physical order (row-major).
+///   When logical and physical layout are equal, the permutation will
+///   always be `([0, 1, .., N-1])` and can therefore be left out.
+///
+/// Description of the serialization:
+/// The metadata must be a valid JSON object including shape of the
+/// contained tensors as an array with key `shape` plus optional
+/// dimension names with keys `dim_names` and ordering of the
+/// dimensions with key `permutation`.
+/// Example: `{ "shape": [2, 5]}`
+/// Example with `dim_names` metadata for NCHW ordered data:
+/// `{ "shape": [100, 200, 500], "dim_names": ["C", "H", "W"]}`
+/// Example of permuted 3-dimensional tensor:
+/// `{ "shape": [100, 200, 500], "permutation": [2, 0, 1]}`
+///
+/// This is the physical layout shape and the shape of the logical layout
+/// would in this case be `[500, 100, 200]`.
+///
+/// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#fixed-shape-tensor>
+#[derive(Debug, Clone, PartialEq)]
+pub struct FixedShapeTensor {
+    /// The data type of individual tensor elements.
+    value_type: DataType,
+
+    /// The metadata of this extension type.
+    metadata: FixedShapeTensorMetadata,
+}
+
+impl FixedShapeTensor {
+    /// Returns a new fixed shape tensor extension type.
+    ///
+    /// # Error
+    ///
+    /// Return an error if the provided dimension names or permutations are
+    /// invalid.
+    pub fn try_new(
+        value_type: DataType,
+        shape: impl IntoIterator<Item = usize>,
+        dimension_names: Option<Vec<String>>,
+        permutations: Option<Vec<usize>>,
+    ) -> Result<Self, ArrowError> {
+        // TODO: are all data types are suitable as value type?
+        FixedShapeTensorMetadata::try_new(shape, dimension_names, permutations).map(|metadata| {
+            Self {
+                value_type,
+                metadata,
+            }
+        })
+    }
+
+    /// Returns the value type of the individual tensor elements.
+    pub fn value_type(&self) -> &DataType {
+        &self.value_type
+    }
+
+    /// Returns the product of all the elements in tensor shape.
+    pub fn list_size(&self) -> usize {
+        self.metadata.list_size()
+    }
+
+    /// Returns the number of dimensions in this fixed shape tensor.
+    pub fn dimensions(&self) -> usize {
+        self.metadata.dimensions()
+    }
+
+    /// Returns the names of the dimensions in this fixed shape tensor, if
+    /// set.
+    pub fn dimension_names(&self) -> Option<&[String]> {
+        self.metadata.dimension_names()
+    }
+
+    /// Returns the indices of the desired ordering of the original
+    /// dimensions, if set.
+    pub fn permutations(&self) -> Option<&[usize]> {
+        self.metadata.permutations()
+    }
+}
+
+/// Extension type metadata for [`FixedShapeTensor`].
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct FixedShapeTensorMetadata {
+    /// The physical shape of the contained tensors.
+    shape: Vec<usize>,
+
+    /// Explicit names to tensor dimensions.
+    dim_names: Option<Vec<String>>,
+
+    /// Indices of the desired ordering of the original dimensions.
+    permutations: Option<Vec<usize>>,
+}
+
+impl FixedShapeTensorMetadata {
+    /// Returns metadata for a fixed shape tensor extension type.
+    ///
+    /// # Error
+    ///
+    /// Return an error if the provided dimension names or permutations are
+    /// invalid.
+    pub fn try_new(
+        shape: impl IntoIterator<Item = usize>,
+        dimension_names: Option<Vec<String>>,
+        permutations: Option<Vec<usize>>,
+    ) -> Result<Self, ArrowError> {
+        let shape = shape.into_iter().collect::<Vec<_>>();
+        let dimensions = shape.len();
+
+        let dim_names = dimension_names.map(|dimension_names| {
+            if dimension_names.len() != dimensions {
+                Err(ArrowError::InvalidArgumentError(format!(
+                    "FixedShapeTensor dimension names size mismatch, expected {dimensions}, found {}", dimension_names.len()
+                )))
+            } else {
+                Ok(dimension_names)
+            }
+        }).transpose()?;
+
+        let permutations = permutations
+            .map(|permutations| {
+                if permutations.len() != dimensions {
+                    Err(ArrowError::InvalidArgumentError(format!(
+                        "FixedShapeTensor permutations size mismatch, expected {dimensions}, found {}",
+                        permutations.len()
+                    )))
+                } else {
+                    let mut sorted_permutations = permutations.clone();
+                    sorted_permutations.sort_unstable();
+                    if (0..dimensions).zip(sorted_permutations).any(|(a, b)| a != b) {
+                        Err(ArrowError::InvalidArgumentError(format!(
+                            "FixedShapeTensor permutations invalid, expected a permutation of [0, 1, .., N-1], where N is the number of dimensions: {dimensions}"
+                        )))
+                    } else {
+                        Ok(permutations)
+                    }
+                }
+            })
+            .transpose()?;
+
+        Ok(Self {
+            shape,
+            dim_names,
+            permutations,
+        })
+    }
+
+    /// Returns the product of all the elements in tensor shape.
+    pub fn list_size(&self) -> usize {
+        self.shape.iter().product()
+    }
+
+    /// Returns the number of dimensions in this fixed shape tensor.
+    pub fn dimensions(&self) -> usize {
+        self.shape.len()
+    }
+
+    /// Returns the names of the dimensions in this fixed shape tensor, if
+    /// set.
+    pub fn dimension_names(&self) -> Option<&[String]> {
+        self.dim_names.as_ref().map(AsRef::as_ref)
+    }
+
+    /// Returns the indices of the desired ordering of the original
+    /// dimensions, if set.
+    pub fn permutations(&self) -> Option<&[usize]> {
+        self.permutations.as_ref().map(AsRef::as_ref)
+    }
+}
+
+impl ExtensionType for FixedShapeTensor {
+    const NAME: &'static str = "arrow.fixed_shape_tensor";
+
+    type Metadata = FixedShapeTensorMetadata;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &self.metadata
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        Some(serde_json::to_string(&self.metadata).expect("metadata serialization"))
+    }
+
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+        metadata.map_or_else(
+            || {
+                Err(ArrowError::InvalidArgumentError(
+                    "FixedShapeTensor extension types requires metadata".to_owned(),
+                ))
+            },
+            |value| {
+                serde_json::from_str(value).map_err(|e| {
+                    ArrowError::InvalidArgumentError(format!(
+                        "FixedShapeTensor metadata deserialization failed: {e}"
+                    ))
+                })
+            },
+        )
+    }
+
+    fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError> {
+        let expected = DataType::new_fixed_size_list(
+            self.value_type.clone(),
+            i32::try_from(self.list_size()).expect("overflow"),
+            false,
+        );
+        data_type
+            .equals_datatype(&expected)
+            .then_some(())
+            .ok_or_else(|| {
+                ArrowError::InvalidArgumentError(format!(
+                    "FixedShapeTensor data type mismatch, expected {expected}, found {data_type}"
+                ))
+            })
+    }
+
+    fn try_new(data_type: &DataType, metadata: Self::Metadata) -> Result<Self, ArrowError> {
+        match data_type {
+            DataType::FixedSizeList(field, list_size) if !field.is_nullable() => {
+                // Make sure the metadata is valid.
+                let metadata = FixedShapeTensorMetadata::try_new(
+                    metadata.shape,
+                    metadata.dim_names,
+                    metadata.permutations,
+                )?;
+                // Make sure it is compatible with this data type.
+                let expected_size = i32::try_from(metadata.list_size()).expect("overflow");
+                if *list_size != expected_size {
+                    Err(ArrowError::InvalidArgumentError(format!(
+                        "FixedShapeTensor list size mismatch, expected {expected_size} (metadata), found {list_size} (data type)"
+                    )))
+                } else {
+                    Ok(Self {
+                        value_type: field.data_type().clone(),
+                        metadata,
+                    })
+                }
+            }
+            data_type => Err(ArrowError::InvalidArgumentError(format!(
+                "FixedShapeTensor data type mismatch, expected FixedSizeList with non-nullable field, found {data_type}"
+            ))),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "canonical_extension_types")]
+    use crate::extension::CanonicalExtensionType;
+    use crate::{
+        extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+        Field,
+    };
+
+    use super::*;
+
+    #[test]
+    fn valid() -> Result<(), ArrowError> {
+        let fixed_shape_tensor = FixedShapeTensor::try_new(
+            DataType::Float32,
+            [100, 200, 500],
+            Some(vec!["C".to_owned(), "H".to_owned(), "W".to_owned()]),
+            Some(vec![2, 0, 1]),
+        )?;
+        let mut field = Field::new_fixed_size_list(
+            "",
+            Field::new("", DataType::Float32, false),
+            i32::try_from(fixed_shape_tensor.list_size()).expect("overflow"),
+            false,
+        );
+        field.try_with_extension_type(fixed_shape_tensor.clone())?;
+        assert_eq!(
+            field.try_extension_type::<FixedShapeTensor>()?,
+            fixed_shape_tensor
+        );
+        #[cfg(feature = "canonical_extension_types")]
+        assert_eq!(
+            field.try_canonical_extension_type()?,
+            CanonicalExtensionType::FixedShapeTensor(fixed_shape_tensor)
+        );
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "Field extension type name missing")]
+    fn missing_name() {
+        let field =
+            Field::new_fixed_size_list("", Field::new("", DataType::Float32, false), 3, false)
+                .with_metadata(
+                    [(
+                        EXTENSION_TYPE_METADATA_KEY.to_owned(),
+                        r#"{ "shape": [100, 200, 500], }"#.to_owned(),
+                    )]
+                    .into_iter()
+                    .collect(),
+                );
+        field.extension_type::<FixedShapeTensor>();
+    }
+
+    #[test]
+    #[should_panic(expected = "FixedShapeTensor data type mismatch, expected FixedSizeList")]
+    fn invalid_type() {
+        let fixed_shape_tensor =
+            FixedShapeTensor::try_new(DataType::Int32, [100, 200, 500], None, None).unwrap();
+        let field = Field::new_fixed_size_list(
+            "",
+            Field::new("", DataType::Float32, false),
+            i32::try_from(fixed_shape_tensor.list_size()).expect("overflow"),
+            false,
+        );
+        field.with_extension_type(fixed_shape_tensor);
+    }
+
+    #[test]
+    #[should_panic(expected = "FixedShapeTensor extension types requires metadata")]
+    fn missing_metadata() {
+        let field =
+            Field::new_fixed_size_list("", Field::new("", DataType::Float32, false), 3, false)
+                .with_metadata(
+                    [(
+                        EXTENSION_TYPE_NAME_KEY.to_owned(),
+                        FixedShapeTensor::NAME.to_owned(),
+                    )]
+                    .into_iter()
+                    .collect(),
+                );
+        field.extension_type::<FixedShapeTensor>();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "FixedShapeTensor metadata deserialization failed: missing field `shape`"
+    )]
+    fn invalid_metadata() {
+        let fixed_shape_tensor =
+            FixedShapeTensor::try_new(DataType::Float32, [100, 200, 500], None, None).unwrap();
+        let field = Field::new_fixed_size_list(
+            "",
+            Field::new("", DataType::Float32, false),
+            i32::try_from(fixed_shape_tensor.list_size()).expect("overflow"),
+            false,
+        )
+        .with_metadata(
+            [
+                (
+                    EXTENSION_TYPE_NAME_KEY.to_owned(),
+                    FixedShapeTensor::NAME.to_owned(),
+                ),
+                (
+                    EXTENSION_TYPE_METADATA_KEY.to_owned(),
+                    r#"{ "not-shape": [] }"#.to_owned(),
+                ),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<FixedShapeTensor>();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "FixedShapeTensor dimension names size mismatch, expected 3, found 2"
+    )]
+    fn invalid_metadata_dimension_names() {
+        FixedShapeTensor::try_new(
+            DataType::Float32,
+            [100, 200, 500],
+            Some(vec!["a".to_owned(), "b".to_owned()]),
+            None,
+        )
+        .unwrap();
+    }
+
+    #[test]
+    #[should_panic(expected = "FixedShapeTensor permutations size mismatch, expected 3, found 2")]
+    fn invalid_metadata_permutations_len() {
+        FixedShapeTensor::try_new(DataType::Float32, [100, 200, 500], None, Some(vec![1, 0]))
+            .unwrap();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "FixedShapeTensor permutations invalid, expected a permutation of [0, 1, .., N-1], where N is the number of dimensions: 3"
+    )]
+    fn invalid_metadata_permutations_values() {
+        FixedShapeTensor::try_new(
+            DataType::Float32,
+            [100, 200, 500],
+            None,
+            Some(vec![4, 3, 2]),
+        )
+        .unwrap();
+    }
+}
diff --git a/arrow-schema/src/extension/canonical/json.rs b/arrow-schema/src/extension/canonical/json.rs
new file mode 100644
index 000000000..0a8a1ae7e
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/json.rs
@@ -0,0 +1,198 @@
+// 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.
+
+//! JSON
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#json>
+
+use serde::{Deserialize, Serialize};
+
+use crate::{extension::ExtensionType, ArrowError, DataType};
+
+/// The extension type for `JSON`.
+///
+/// Extension name: `arrow.json`.
+///
+/// The storage type of this extension is `String` or `LargeString` or
+/// `StringView`. Only UTF-8 encoded JSON as specified in [rfc8259](https://datatracker.ietf.org/doc/html/rfc8259)
+/// is supported.
+///
+/// This type does not have any parameters.
+///
+/// Metadata is either an empty string or a JSON string with an empty
+/// object. In the future, additional fields may be added, but they are not
+/// required to interpret the array.
+///
+/// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#json>
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct Json(JsonMetadata);
+
+/// Empty object
+#[derive(Debug, Clone, Copy, PartialEq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+struct Empty {}
+
+/// Extension type metadata for [`Json`].
+#[derive(Debug, Default, Clone, PartialEq)]
+pub struct JsonMetadata(Option<Empty>);
+
+impl ExtensionType for Json {
+    const NAME: &'static str = "arrow.json";
+
+    type Metadata = JsonMetadata;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &self.0
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        Some(
+            self.metadata()
+                .0
+                .as_ref()
+                .map(serde_json::to_string)
+                .map(Result::unwrap)
+                .unwrap_or_else(|| "".to_owned()),
+        )
+    }
+
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+        const ERR: &str = "Json extension type metadata is either an empty string or a JSON string with an empty object";
+        metadata
+            .map_or_else(
+                || Err(ArrowError::InvalidArgumentError(ERR.to_owned())),
+                |metadata| {
+                    match metadata {
+                        // Empty string
+                        "" => Ok(None),
+                        value => serde_json::from_str::<Empty>(value)
+                            .map(Option::Some)
+                            .map_err(|_| ArrowError::InvalidArgumentError(ERR.to_owned())),
+                    }
+                },
+            )
+            .map(JsonMetadata)
+    }
+
+    fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError> {
+        match data_type {
+            DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Ok(()),
+            data_type => Err(ArrowError::InvalidArgumentError(format!(
+                "Json data type mismatch, expected one of Utf8, LargeUtf8, Utf8View, found {data_type}"
+            ))),
+        }
+    }
+
+    fn try_new(data_type: &DataType, metadata: Self::Metadata) -> Result<Self, ArrowError> {
+        let json = Self(metadata);
+        json.supports_data_type(data_type)?;
+        Ok(json)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "canonical_extension_types")]
+    use crate::extension::CanonicalExtensionType;
+    use crate::{
+        extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+        Field,
+    };
+
+    use super::*;
+
+    #[test]
+    fn valid() -> Result<(), ArrowError> {
+        let mut field = Field::new("", DataType::Utf8, false);
+        field.try_with_extension_type(Json::default())?;
+        assert_eq!(
+            field.metadata().get(EXTENSION_TYPE_METADATA_KEY),
+            Some(&"".to_owned())
+        );
+        assert_eq!(
+            field.try_extension_type::<Json>()?,
+            Json(JsonMetadata(None))
+        );
+
+        let mut field = Field::new("", DataType::LargeUtf8, false);
+        field.try_with_extension_type(Json(JsonMetadata(Some(Empty {}))))?;
+        assert_eq!(
+            field.metadata().get(EXTENSION_TYPE_METADATA_KEY),
+            Some(&"{}".to_owned())
+        );
+        assert_eq!(
+            field.try_extension_type::<Json>()?,
+            Json(JsonMetadata(Some(Empty {})))
+        );
+
+        let mut field = Field::new("", DataType::Utf8View, false);
+        field.try_with_extension_type(Json::default())?;
+        field.try_extension_type::<Json>()?;
+        #[cfg(feature = "canonical_extension_types")]
+        assert_eq!(
+            field.try_canonical_extension_type()?,
+            CanonicalExtensionType::Json(Json::default())
+        );
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "Field extension type name missing")]
+    fn missing_name() {
+        let field = Field::new("", DataType::Int8, false).with_metadata(
+            [(EXTENSION_TYPE_METADATA_KEY.to_owned(), "{}".to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        field.extension_type::<Json>();
+    }
+
+    #[test]
+    #[should_panic(expected = "expected one of Utf8, LargeUtf8, Utf8View, found Null")]
+    fn invalid_type() {
+        Field::new("", DataType::Null, false).with_extension_type(Json::default());
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "Json extension type metadata is either an empty string or a JSON string with an empty object"
+    )]
+    fn invalid_metadata() {
+        let field = Field::new("", DataType::Utf8, false).with_metadata(
+            [
+                (EXTENSION_TYPE_NAME_KEY.to_owned(), Json::NAME.to_owned()),
+                (EXTENSION_TYPE_METADATA_KEY.to_owned(), "1234".to_owned()),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<Json>();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "Json extension type metadata is either an empty string or a JSON string with an empty object"
+    )]
+    fn missing_metadata() {
+        let field = Field::new("", DataType::LargeUtf8, false).with_metadata(
+            [(EXTENSION_TYPE_NAME_KEY.to_owned(), Json::NAME.to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        field.extension_type::<Json>();
+    }
+}
diff --git a/arrow-schema/src/extension/canonical/mod.rs b/arrow-schema/src/extension/canonical/mod.rs
new file mode 100644
index 000000000..3d66299ca
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/mod.rs
@@ -0,0 +1,142 @@
+// 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.
+
+//! Canonical extension types.
+//!
+//! The Arrow columnar format allows defining extension types so as to extend
+//! standard Arrow data types with custom semantics. Often these semantics will
+//! be specific to a system or application. However, it is beneficial to share
+//! the definitions of well-known extension types so as to improve
+//! interoperability between different systems integrating Arrow columnar data.
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#format-canonical-extensions>
+
+mod bool8;
+pub use bool8::Bool8;
+mod fixed_shape_tensor;
+pub use fixed_shape_tensor::{FixedShapeTensor, FixedShapeTensorMetadata};
+mod json;
+pub use json::{Json, JsonMetadata};
+mod opaque;
+pub use opaque::{Opaque, OpaqueMetadata};
+mod uuid;
+pub use uuid::Uuid;
+mod variable_shape_tensor;
+pub use variable_shape_tensor::{VariableShapeTensor, VariableShapeTensorMetadata};
+
+use crate::{ArrowError, Field};
+
+use super::ExtensionType;
+
+/// Canonical extension types.
+///
+/// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#format-canonical-extensions>
+#[non_exhaustive]
+#[derive(Debug, Clone, PartialEq)]
+pub enum CanonicalExtensionType {
+    /// The extension type for `FixedShapeTensor`.
+    ///
+    /// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#fixed-shape-tensor>
+    FixedShapeTensor(FixedShapeTensor),
+
+    /// The extension type for `VariableShapeTensor`.
+    ///
+    /// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#variable-shape-tensor>
+    VariableShapeTensor(VariableShapeTensor),
+
+    /// The extension type for 'JSON'.
+    ///
+    /// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#json>
+    Json(Json),
+
+    /// The extension type for `UUID`.
+    ///
+    /// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#uuid>
+    Uuid(Uuid),
+
+    /// The extension type for `Opaque`.
+    ///
+    /// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#opaque>
+    Opaque(Opaque),
+
+    /// The extension type for `Bool8`.
+    ///
+    /// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#bit-boolean>
+    Bool8(Bool8),
+}
+
+impl TryFrom<&Field> for CanonicalExtensionType {
+    type Error = ArrowError;
+
+    fn try_from(value: &Field) -> Result<Self, Self::Error> {
+        // Canonical extension type names start with `arrow.`
+        match value.extension_type_name() {
+            // An extension type name with an `arrow.` prefix
+            Some(name) if name.starts_with("arrow.") => match name {
+                FixedShapeTensor::NAME => value.try_extension_type::<FixedShapeTensor>().map(Into::into),
+                VariableShapeTensor::NAME => value.try_extension_type::<VariableShapeTensor>().map(Into::into),
+                Json::NAME => value.try_extension_type::<Json>().map(Into::into),
+                Uuid::NAME => value.try_extension_type::<Uuid>().map(Into::into),
+                Opaque::NAME => value.try_extension_type::<Opaque>().map(Into::into),
+                Bool8::NAME => value.try_extension_type::<Bool8>().map(Into::into),
+                _ => Err(ArrowError::InvalidArgumentError(format!("Unsupported canonical extension type: {name}"))),
+            },
+            // Name missing the expected prefix
+            Some(name) => Err(ArrowError::InvalidArgumentError(format!(
+                "Field extension type name mismatch, expected a name with an `arrow.` prefix, found {name}"
+            ))),
+            // Name missing
+            None => Err(ArrowError::InvalidArgumentError("Field extension type name missing".to_owned())),
+        }
+    }
+}
+
+impl From<FixedShapeTensor> for CanonicalExtensionType {
+    fn from(value: FixedShapeTensor) -> Self {
+        CanonicalExtensionType::FixedShapeTensor(value)
+    }
+}
+
+impl From<VariableShapeTensor> for CanonicalExtensionType {
+    fn from(value: VariableShapeTensor) -> Self {
+        CanonicalExtensionType::VariableShapeTensor(value)
+    }
+}
+
+impl From<Json> for CanonicalExtensionType {
+    fn from(value: Json) -> Self {
+        CanonicalExtensionType::Json(value)
+    }
+}
+
+impl From<Uuid> for CanonicalExtensionType {
+    fn from(value: Uuid) -> Self {
+        CanonicalExtensionType::Uuid(value)
+    }
+}
+
+impl From<Opaque> for CanonicalExtensionType {
+    fn from(value: Opaque) -> Self {
+        CanonicalExtensionType::Opaque(value)
+    }
+}
+
+impl From<Bool8> for CanonicalExtensionType {
+    fn from(value: Bool8) -> Self {
+        CanonicalExtensionType::Bool8(value)
+    }
+}
diff --git a/arrow-schema/src/extension/canonical/opaque.rs b/arrow-schema/src/extension/canonical/opaque.rs
new file mode 100644
index 000000000..1db7265cf
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/opaque.rs
@@ -0,0 +1,201 @@
+// 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.
+
+//! Opaque
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#opaque>
+
+use serde::{Deserialize, Serialize};
+
+use crate::{extension::ExtensionType, ArrowError, DataType};
+
+/// The extension type for `Opaque`.
+///
+/// Extension name: `arrow.opaque`.
+///
+/// Opaque represents a type that an Arrow-based system received from an
+/// external (often non-Arrow) system, but that it cannot interpret. In this
+/// case, it can pass on Opaque to its clients to at least show that a field
+/// exists and preserve metadata about the type from the other system.
+///
+/// The storage type of this extension is any type. If there is no underlying
+/// data, the storage type should be Null.
+#[derive(Debug, Clone, PartialEq)]
+pub struct Opaque(OpaqueMetadata);
+
+impl Opaque {
+    /// Returns a new `Opaque` extension type.
+    pub fn new(type_name: impl Into<String>, vendor_name: impl Into<String>) -> Self {
+        Self(OpaqueMetadata::new(type_name, vendor_name))
+    }
+
+    /// Returns the name of the unknown type in the external system.
+    pub fn type_name(&self) -> &str {
+        self.0.type_name()
+    }
+
+    /// Returns the name of the external system.
+    pub fn vendor_name(&self) -> &str {
+        self.0.vendor_name()
+    }
+}
+
+impl From<OpaqueMetadata> for Opaque {
+    fn from(value: OpaqueMetadata) -> Self {
+        Self(value)
+    }
+}
+
+/// Extension type metadata for [`Opaque`].
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct OpaqueMetadata {
+    /// Name of the unknown type in the external system.
+    type_name: String,
+
+    /// Name of the external system.
+    vendor_name: String,
+}
+
+impl OpaqueMetadata {
+    /// Returns a new `OpaqueMetadata`.
+    pub fn new(type_name: impl Into<String>, vendor_name: impl Into<String>) -> Self {
+        OpaqueMetadata {
+            type_name: type_name.into(),
+            vendor_name: vendor_name.into(),
+        }
+    }
+
+    /// Returns the name of the unknown type in the external system.
+    pub fn type_name(&self) -> &str {
+        &self.type_name
+    }
+
+    /// Returns the name of the external system.
+    pub fn vendor_name(&self) -> &str {
+        &self.vendor_name
+    }
+}
+
+impl ExtensionType for Opaque {
+    const NAME: &'static str = "arrow.opaque";
+
+    type Metadata = OpaqueMetadata;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &self.0
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        Some(serde_json::to_string(self.metadata()).expect("metadata serialization"))
+    }
+
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+        metadata.map_or_else(
+            || {
+                Err(ArrowError::InvalidArgumentError(
+                    "Opaque extension types requires metadata".to_owned(),
+                ))
+            },
+            |value| {
+                serde_json::from_str(value).map_err(|e| {
+                    ArrowError::InvalidArgumentError(format!(
+                        "Opaque metadata deserialization failed: {e}"
+                    ))
+                })
+            },
+        )
+    }
+
+    fn supports_data_type(&self, _data_type: &DataType) -> Result<(), ArrowError> {
+        // Any type
+        Ok(())
+    }
+
+    fn try_new(_data_type: &DataType, metadata: Self::Metadata) -> Result<Self, ArrowError> {
+        Ok(Self::from(metadata))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "canonical_extension_types")]
+    use crate::extension::CanonicalExtensionType;
+    use crate::{
+        extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+        Field,
+    };
+
+    use super::*;
+
+    #[test]
+    fn valid() -> Result<(), ArrowError> {
+        let opaque = Opaque::new("name", "vendor");
+        let mut field = Field::new("", DataType::Null, false);
+        field.try_with_extension_type(opaque.clone())?;
+        assert_eq!(field.try_extension_type::<Opaque>()?, opaque);
+        #[cfg(feature = "canonical_extension_types")]
+        assert_eq!(
+            field.try_canonical_extension_type()?,
+            CanonicalExtensionType::Opaque(opaque)
+        );
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "Field extension type name missing")]
+    fn missing_name() {
+        let field = Field::new("", DataType::Null, false).with_metadata(
+            [(
+                EXTENSION_TYPE_METADATA_KEY.to_owned(),
+                r#"{ "type_name": "type", "vendor_name": "vendor" }"#.to_owned(),
+            )]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<Opaque>();
+    }
+
+    #[test]
+    #[should_panic(expected = "Opaque extension types requires metadata")]
+    fn missing_metadata() {
+        let field = Field::new("", DataType::Null, false).with_metadata(
+            [(EXTENSION_TYPE_NAME_KEY.to_owned(), Opaque::NAME.to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        field.extension_type::<Opaque>();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "Opaque metadata deserialization failed: missing field `vendor_name`"
+    )]
+    fn invalid_metadata() {
+        let field = Field::new("", DataType::Null, false).with_metadata(
+            [
+                (EXTENSION_TYPE_NAME_KEY.to_owned(), Opaque::NAME.to_owned()),
+                (
+                    EXTENSION_TYPE_METADATA_KEY.to_owned(),
+                    r#"{ "type_name": "no-vendor" }"#.to_owned(),
+                ),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<Opaque>();
+    }
+}
diff --git a/arrow-schema/src/extension/canonical/uuid.rs b/arrow-schema/src/extension/canonical/uuid.rs
new file mode 100644
index 000000000..8b2e71b7b
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/uuid.rs
@@ -0,0 +1,128 @@
+// 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.
+
+//! UUID
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#uuid>
+
+use crate::{extension::ExtensionType, ArrowError, DataType};
+
+/// The extension type for `UUID`.
+///
+/// Extension name: `arrow.uuid`.
+///
+/// The storage type of the extension is `FixedSizeBinary` with a length of
+/// 16 bytes.
+///
+/// Note:
+/// A specific UUID version is not required or guaranteed. This extension
+/// represents UUIDs as `FixedSizeBinary(16)` with big-endian notation and
+/// does not interpret the bytes in any way.
+///
+/// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#uuid>
+#[derive(Debug, Default, Clone, Copy, PartialEq)]
+pub struct Uuid;
+
+impl ExtensionType for Uuid {
+    const NAME: &'static str = "arrow.uuid";
+
+    type Metadata = ();
+
+    fn metadata(&self) -> &Self::Metadata {
+        &()
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        None
+    }
+
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+        metadata.map_or_else(
+            || Ok(()),
+            |_| {
+                Err(ArrowError::InvalidArgumentError(
+                    "Uuid extension type expects no metadata".to_owned(),
+                ))
+            },
+        )
+    }
+
+    fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError> {
+        match data_type {
+            DataType::FixedSizeBinary(16) => Ok(()),
+            data_type => Err(ArrowError::InvalidArgumentError(format!(
+                "Uuid data type mismatch, expected FixedSizeBinary(16), found {data_type}"
+            ))),
+        }
+    }
+
+    fn try_new(data_type: &DataType, _metadata: Self::Metadata) -> Result<Self, ArrowError> {
+        Self.supports_data_type(data_type).map(|_| Self)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "canonical_extension_types")]
+    use crate::extension::CanonicalExtensionType;
+    use crate::{
+        extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+        Field,
+    };
+
+    use super::*;
+
+    #[test]
+    fn valid() -> Result<(), ArrowError> {
+        let mut field = Field::new("", DataType::FixedSizeBinary(16), false);
+        field.try_with_extension_type(Uuid)?;
+        field.try_extension_type::<Uuid>()?;
+        #[cfg(feature = "canonical_extension_types")]
+        assert_eq!(
+            field.try_canonical_extension_type()?,
+            CanonicalExtensionType::Uuid(Uuid)
+        );
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "Field extension type name missing")]
+    fn missing_name() {
+        let field = Field::new("", DataType::FixedSizeBinary(16), false);
+        field.extension_type::<Uuid>();
+    }
+
+    #[test]
+    #[should_panic(expected = "expected FixedSizeBinary(16), found FixedSizeBinary(8)")]
+    fn invalid_type() {
+        Field::new("", DataType::FixedSizeBinary(8), false).with_extension_type(Uuid);
+    }
+
+    #[test]
+    #[should_panic(expected = "Uuid extension type expects no metadata")]
+    fn with_metadata() {
+        let field = Field::new("", DataType::FixedSizeBinary(16), false).with_metadata(
+            [
+                (EXTENSION_TYPE_NAME_KEY.to_owned(), Uuid::NAME.to_owned()),
+                (EXTENSION_TYPE_METADATA_KEY.to_owned(), "".to_owned()),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<Uuid>();
+    }
+}
diff --git a/arrow-schema/src/extension/canonical/variable_shape_tensor.rs b/arrow-schema/src/extension/canonical/variable_shape_tensor.rs
new file mode 100644
index 000000000..804591776
--- /dev/null
+++ b/arrow-schema/src/extension/canonical/variable_shape_tensor.rs
@@ -0,0 +1,551 @@
+// 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.
+
+//! VariableShapeTensor
+//!
+//! <https://arrow.apache.org/docs/format/CanonicalExtensions.html#variable-shape-tensor>
+
+use serde::{Deserialize, Serialize};
+
+use crate::{extension::ExtensionType, ArrowError, DataType, Field};
+
+/// The extension type for `VariableShapeTensor`.
+///
+/// Extension name: `arrow.variable_shape_tensor`.
+///
+/// The storage type of the extension is: StructArray where struct is composed
+/// of data and shape fields describing a single tensor per row:
+/// - `data` is a List holding tensor elements (each list element is a single
+///   tensor). The List’s value type is the value type of the tensor, such as
+///   an integer or floating-point type.
+/// - `shape` is a `FixedSizeList<int32>[ndim]` of the tensor shape where the
+///   size of the list `ndim` is equal to the number of dimensions of the
+///   tensor.
+///
+/// Extension type parameters:
+/// `value_type`: the Arrow data type of individual tensor elements.
+///
+/// Optional parameters describing the logical layout:
+/// - `dim_names`: explicit names to tensor dimensions as an array. The length
+///   of it should be equal to the shape length and equal to the number of
+///   dimensions.
+///   `dim_names` can be used if the dimensions have well-known names and they
+///   map to the physical layout (row-major).
+/// - `permutation`: indices of the desired ordering of the original
+///   dimensions, defined as an array.
+///   The indices contain a permutation of the values `[0, 1, .., N-1]` where
+///   `N` is the number of dimensions. The permutation indicates which
+///   dimension of the logical layout corresponds to which dimension of the
+///   physical tensor (the i-th dimension of the logical view corresponds to
+///   the dimension with number `permutations[i]` of the physical tensor).
+///   Permutation can be useful in case the logical order of the tensor is a
+///   permutation of the physical order (row-major).
+///   When logical and physical layout are equal, the permutation will always
+///   be (`[0, 1, .., N-1]`) and can therefore be left out.
+/// - `uniform_shape`: sizes of individual tensor’s dimensions which are
+///   guaranteed to stay constant in uniform dimensions and can vary in non-
+///   uniform dimensions. This holds over all tensors in the array. Sizes in
+///   uniform dimensions are represented with int32 values, while sizes of the
+///   non-uniform dimensions are not known in advance and are represented with
+///   null. If `uniform_shape` is not provided it is assumed that all
+///   dimensions are non-uniform. An array containing a tensor with shape (2,
+///   3, 4) and whose first and last dimensions are uniform would have
+///   `uniform_shape` (2, null, 4). This allows for interpreting the tensor
+///   correctly without accounting for uniform dimensions while still
+///   permitting optional optimizations that take advantage of the uniformity.
+///
+/// <https://arrow.apache.org/docs/format/CanonicalExtensions.html#variable-shape-tensor>
+#[derive(Debug, Clone, PartialEq)]
+pub struct VariableShapeTensor {
+    /// The data type of individual tensor elements.
+    value_type: DataType,
+
+    /// The number of dimensions of the tensor.
+    dimensions: usize,
+
+    /// The metadata of this extension type.
+    metadata: VariableShapeTensorMetadata,
+}
+
+impl VariableShapeTensor {
+    /// Returns a new variable shape tensor extension type.
+    ///
+    /// # Error
+    ///
+    /// Return an error if the provided dimension names, permutations or
+    /// uniform shapes are invalid.
+    pub fn try_new(
+        value_type: DataType,
+        dimensions: usize,
+        dimension_names: Option<Vec<String>>,
+        permutations: Option<Vec<usize>>,
+        uniform_shapes: Option<Vec<Option<i32>>>,
+    ) -> Result<Self, ArrowError> {
+        // TODO: are all data types are suitable as value type?
+        VariableShapeTensorMetadata::try_new(
+            dimensions,
+            dimension_names,
+            permutations,
+            uniform_shapes,
+        )
+        .map(|metadata| Self {
+            value_type,
+            dimensions,
+            metadata,
+        })
+    }
+
+    /// Returns the value type of the individual tensor elements.
+    pub fn value_type(&self) -> &DataType {
+        &self.value_type
+    }
+
+    /// Returns the number of dimensions  in this variable shape tensor.
+    pub fn dimensions(&self) -> usize {
+        self.dimensions
+    }
+
+    /// Returns the names of the dimensions in this variable shape tensor, if
+    /// set.
+    pub fn dimension_names(&self) -> Option<&[String]> {
+        self.metadata.dimension_names()
+    }
+
+    /// Returns the indices of the desired ordering of the original
+    /// dimensions, if set.
+    pub fn permutations(&self) -> Option<&[usize]> {
+        self.metadata.permutations()
+    }
+
+    /// Returns sizes of individual tensor’s dimensions which are guaranteed
+    /// to stay constant in uniform dimensions and can vary in non-uniform
+    /// dimensions.
+    pub fn uniform_shapes(&self) -> Option<&[Option<i32>]> {
+        self.metadata.uniform_shapes()
+    }
+}
+
+/// Extension type metadata for [`VariableShapeTensor`].
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct VariableShapeTensorMetadata {
+    /// Explicit names to tensor dimensions.
+    dim_names: Option<Vec<String>>,
+
+    /// Indices of the desired ordering of the original dimensions.
+    permutations: Option<Vec<usize>>,
+
+    /// Sizes of individual tensor’s dimensions which are guaranteed to stay
+    /// constant in uniform dimensions and can vary in non-uniform dimensions.
+    uniform_shape: Option<Vec<Option<i32>>>,
+}
+
+impl VariableShapeTensorMetadata {
+    /// Returns metadata for a variable shape tensor extension type.
+    ///
+    /// # Error
+    ///
+    /// Return an error if the provided dimension names, permutations or
+    /// uniform shapes are invalid.
+    pub fn try_new(
+        dimensions: usize,
+        dimension_names: Option<Vec<String>>,
+        permutations: Option<Vec<usize>>,
+        uniform_shapes: Option<Vec<Option<i32>>>,
+    ) -> Result<Self, ArrowError> {
+        let dim_names = dimension_names.map(|dimension_names| {
+            if dimension_names.len() != dimensions {
+                Err(ArrowError::InvalidArgumentError(format!(
+                    "VariableShapeTensor dimension names size mismatch, expected {dimensions}, found {}", dimension_names.len()
+                )))
+            } else {
+                Ok(dimension_names)
+            }
+        }).transpose()?;
+
+        let permutations = permutations
+            .map(|permutations| {
+                if permutations.len() != dimensions {
+                    Err(ArrowError::InvalidArgumentError(format!(
+                        "VariableShapeTensor permutations size mismatch, expected {dimensions}, found {}",
+                        permutations.len()
+                    )))
+                } else {
+                    let mut sorted_permutations = permutations.clone();
+                    sorted_permutations.sort_unstable();
+                    if (0..dimensions).zip(sorted_permutations).any(|(a, b)| a != b) {
+                        Err(ArrowError::InvalidArgumentError(format!(
+                            "VariableShapeTensor permutations invalid, expected a permutation of [0, 1, .., N-1], where N is the number of dimensions: {dimensions}"
+                        )))
+                    } else {
+                        Ok(permutations)
+                    }
+                }
+            })
+            .transpose()?;
+
+        let uniform_shape = uniform_shapes
+            .map(|uniform_shapes| {
+                if uniform_shapes.len() != dimensions {
+                    Err(ArrowError::InvalidArgumentError(format!(
+                        "VariableShapeTensor uniform shapes size mismatch, expected {dimensions}, found {}",
+                        uniform_shapes.len()
+                    )))
+                } else {
+                    Ok(uniform_shapes)
+                }
+            })
+            .transpose()?;
+
+        Ok(Self {
+            dim_names,
+            permutations,
+            uniform_shape,
+        })
+    }
+
+    /// Returns the names of the dimensions in this variable shape tensor, if
+    /// set.
+    pub fn dimension_names(&self) -> Option<&[String]> {
+        self.dim_names.as_ref().map(AsRef::as_ref)
+    }
+
+    /// Returns the indices of the desired ordering of the original dimensions,
+    /// if set.
+    pub fn permutations(&self) -> Option<&[usize]> {
+        self.permutations.as_ref().map(AsRef::as_ref)
+    }
+
+    /// Returns sizes of individual tensor’s dimensions which are guaranteed
+    /// to stay constant in uniform dimensions and can vary in non-uniform
+    /// dimensions.
+    pub fn uniform_shapes(&self) -> Option<&[Option<i32>]> {
+        self.uniform_shape.as_ref().map(AsRef::as_ref)
+    }
+}
+
+impl ExtensionType for VariableShapeTensor {
+    const NAME: &'static str = "arrow.variable_shape_tensor";
+
+    type Metadata = VariableShapeTensorMetadata;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &self.metadata
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        Some(serde_json::to_string(self.metadata()).expect("metadata serialization"))
+    }
+
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+        metadata.map_or_else(
+            || {
+                Err(ArrowError::InvalidArgumentError(
+                    "VariableShapeTensor extension types requires metadata".to_owned(),
+                ))
+            },
+            |value| {
+                serde_json::from_str(value).map_err(|e| {
+                    ArrowError::InvalidArgumentError(format!(
+                        "VariableShapeTensor metadata deserialization failed: {e}"
+                    ))
+                })
+            },
+        )
+    }
+
+    fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError> {
+        let expected = DataType::Struct(
+            [
+                Field::new_list(
+                    "data",
+                    Field::new_list_field(self.value_type.clone(), false),
+                    false,
+                ),
+                Field::new(
+                    "shape",
+                    DataType::new_fixed_size_list(
+                        DataType::Int32,
+                        i32::try_from(self.dimensions()).expect("overflow"),
+                        false,
+                    ),
+                    false,
+                ),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        data_type
+            .equals_datatype(&expected)
+            .then_some(())
+            .ok_or_else(|| {
+                ArrowError::InvalidArgumentError(format!(
+                    "VariableShapeTensor data type mismatch, expected {expected}, found {data_type}"
+                ))
+            })
+    }
+
+    fn try_new(data_type: &DataType, metadata: Self::Metadata) -> Result<Self, ArrowError> {
+        match data_type {
+            DataType::Struct(fields)
+                if fields.len() == 2
+                    && matches!(fields.find("data"), Some((0, _)))
+                    && matches!(fields.find("shape"), Some((1, _))) =>
+            {
+                let shape_field = &fields[1];
+                match shape_field.data_type() {
+                    DataType::FixedSizeList(_, list_size) => {
+                        let dimensions = usize::try_from(*list_size).expect("conversion failed");
+                        // Make sure the metadata is valid.
+                        let metadata = VariableShapeTensorMetadata::try_new(dimensions, metadata.dim_names, metadata.permutations, metadata.uniform_shape)?;
+                        let data_field = &fields[0];
+                        match data_field.data_type() {
+                            DataType::List(field) => {
+                                Ok(Self {
+                                    value_type: field.data_type().clone(),
+                                    dimensions,
+                                    metadata
+                                })
+                            }
+                            data_type => Err(ArrowError::InvalidArgumentError(format!(
+                                "VariableShapeTensor data type mismatch, expected List for data field, found {data_type}"
+                            ))),
+                        }
+                    }
+                    data_type => Err(ArrowError::InvalidArgumentError(format!(
+                        "VariableShapeTensor data type mismatch, expected FixedSizeList for shape field, found {data_type}"
+                    ))),
+                }
+            }
+            data_type => Err(ArrowError::InvalidArgumentError(format!(
+                "VariableShapeTensor data type mismatch, expected Struct with 2 fields (data and shape), found {data_type}"
+            ))),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "canonical_extension_types")]
+    use crate::extension::CanonicalExtensionType;
+    use crate::{
+        extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+        Field,
+    };
+
+    use super::*;
+
+    #[test]
+    fn valid() -> Result<(), ArrowError> {
+        let variable_shape_tensor = VariableShapeTensor::try_new(
+            DataType::Float32,
+            3,
+            Some(vec!["C".to_owned(), "H".to_owned(), "W".to_owned()]),
+            Some(vec![2, 0, 1]),
+            Some(vec![Some(400), None, Some(3)]),
+        )?;
+        let mut field = Field::new_struct(
+            "",
+            vec![
+                Field::new_list(
+                    "data",
+                    Field::new_list_field(DataType::Float32, false),
+                    false,
+                ),
+                Field::new_fixed_size_list(
+                    "shape",
+                    Field::new("", DataType::Int32, false),
+                    3,
+                    false,
+                ),
+            ],
+            false,
+        );
+        field.try_with_extension_type(variable_shape_tensor.clone())?;
+        assert_eq!(
+            field.try_extension_type::<VariableShapeTensor>()?,
+            variable_shape_tensor
+        );
+        #[cfg(feature = "canonical_extension_types")]
+        assert_eq!(
+            field.try_canonical_extension_type()?,
+            CanonicalExtensionType::VariableShapeTensor(variable_shape_tensor)
+        );
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "Field extension type name missing")]
+    fn missing_name() {
+        let field = Field::new_struct(
+            "",
+            vec![
+                Field::new_list(
+                    "data",
+                    Field::new_list_field(DataType::Float32, false),
+                    false,
+                ),
+                Field::new_fixed_size_list(
+                    "shape",
+                    Field::new("", DataType::Int32, false),
+                    3,
+                    false,
+                ),
+            ],
+            false,
+        )
+        .with_metadata(
+            [(EXTENSION_TYPE_METADATA_KEY.to_owned(), "{}".to_owned())]
+                .into_iter()
+                .collect(),
+        );
+        field.extension_type::<VariableShapeTensor>();
+    }
+
+    #[test]
+    #[should_panic(expected = "VariableShapeTensor data type mismatch, expected Struct")]
+    fn invalid_type() {
+        let variable_shape_tensor =
+            VariableShapeTensor::try_new(DataType::Int32, 3, None, None, None).unwrap();
+        let field = Field::new_struct(
+            "",
+            vec![
+                Field::new_list(
+                    "data",
+                    Field::new_list_field(DataType::Float32, false),
+                    false,
+                ),
+                Field::new_fixed_size_list(
+                    "shape",
+                    Field::new("", DataType::Int32, false),
+                    3,
+                    false,
+                ),
+            ],
+            false,
+        );
+        field.with_extension_type(variable_shape_tensor);
+    }
+
+    #[test]
+    #[should_panic(expected = "VariableShapeTensor extension types requires metadata")]
+    fn missing_metadata() {
+        let field = Field::new_struct(
+            "",
+            vec![
+                Field::new_list(
+                    "data",
+                    Field::new_list_field(DataType::Float32, false),
+                    false,
+                ),
+                Field::new_fixed_size_list(
+                    "shape",
+                    Field::new("", DataType::Int32, false),
+                    3,
+                    false,
+                ),
+            ],
+            false,
+        )
+        .with_metadata(
+            [(
+                EXTENSION_TYPE_NAME_KEY.to_owned(),
+                VariableShapeTensor::NAME.to_owned(),
+            )]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<VariableShapeTensor>();
+    }
+
+    #[test]
+    #[should_panic(expected = "VariableShapeTensor metadata deserialization failed: invalid type:")]
+    fn invalid_metadata() {
+        let field = Field::new_struct(
+            "",
+            vec![
+                Field::new_list(
+                    "data",
+                    Field::new_list_field(DataType::Float32, false),
+                    false,
+                ),
+                Field::new_fixed_size_list(
+                    "shape",
+                    Field::new("", DataType::Int32, false),
+                    3,
+                    false,
+                ),
+            ],
+            false,
+        )
+        .with_metadata(
+            [
+                (
+                    EXTENSION_TYPE_NAME_KEY.to_owned(),
+                    VariableShapeTensor::NAME.to_owned(),
+                ),
+                (
+                    EXTENSION_TYPE_METADATA_KEY.to_owned(),
+                    r#"{ "dim_names": [1, null, 3, 4] }"#.to_owned(),
+                ),
+            ]
+            .into_iter()
+            .collect(),
+        );
+        field.extension_type::<VariableShapeTensor>();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "VariableShapeTensor dimension names size mismatch, expected 3, found 2"
+    )]
+    fn invalid_metadata_dimension_names() {
+        VariableShapeTensor::try_new(
+            DataType::Float32,
+            3,
+            Some(vec!["a".to_owned(), "b".to_owned()]),
+            None,
+            None,
+        )
+        .unwrap();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "VariableShapeTensor permutations size mismatch, expected 3, found 2"
+    )]
+    fn invalid_metadata_permutations_len() {
+        VariableShapeTensor::try_new(DataType::Float32, 3, None, Some(vec![1, 0]), None).unwrap();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "VariableShapeTensor permutations invalid, expected a permutation of [0, 1, .., N-1], where N is the number of dimensions: 3"
+    )]
+    fn invalid_metadata_permutations_values() {
+        VariableShapeTensor::try_new(DataType::Float32, 3, None, Some(vec![4, 3, 2]), None)
+            .unwrap();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "VariableShapeTensor uniform shapes size mismatch, expected 3, found 2"
+    )]
+    fn invalid_metadata_uniform_shapes() {
+        VariableShapeTensor::try_new(DataType::Float32, 3, None, None, Some(vec![None, Some(1)]))
+            .unwrap();
+    }
+}
diff --git a/arrow-schema/src/extension/mod.rs b/arrow-schema/src/extension/mod.rs
new file mode 100644
index 000000000..c5119873a
--- /dev/null
+++ b/arrow-schema/src/extension/mod.rs
@@ -0,0 +1,260 @@
+// 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.
+
+//! Extension types.
+//!
+//! <div class="warning">This module is experimental. There might be breaking changes between minor releases.</div>
+
+#[cfg(feature = "canonical_extension_types")]
+mod canonical;
+#[cfg(feature = "canonical_extension_types")]
+pub use canonical::*;
+
+use crate::{ArrowError, DataType};
+
+/// The metadata key for the string name identifying an [`ExtensionType`].
+pub const EXTENSION_TYPE_NAME_KEY: &str = "ARROW:extension:name";
+
+/// The metadata key for a serialized representation of the [`ExtensionType`]
+/// necessary to reconstruct the custom type.
+pub const EXTENSION_TYPE_METADATA_KEY: &str = "ARROW:extension:metadata";
+
+/// Extension types.
+///
+/// User-defined “extension” types can be defined setting certain key value
+/// pairs in the [`Field`] metadata structure. These extension keys are:
+/// - [`EXTENSION_TYPE_NAME_KEY`]
+/// - [`EXTENSION_TYPE_METADATA_KEY`]
+///
+/// Canonical extension types support in this crate requires the
+/// `canonical_extension_types` feature.
+///
+/// Extension types may or may not use the [`EXTENSION_TYPE_METADATA_KEY`]
+/// field.
+///
+/// # Example
+///
+/// The example below demonstrates how to implement this trait for a `Uuid`
+/// type. Note this is not the canonical extension type for `Uuid`, which does
+/// not include information about the `Uuid` version.
+///
+/// ```
+/// # use arrow_schema::ArrowError;
+/// # fn main() -> Result<(), ArrowError> {
+/// use arrow_schema::{DataType, extension::ExtensionType, Field};
+/// use std::{fmt, str::FromStr};
+///
+/// /// The different Uuid versions.
+/// #[derive(Clone, Copy, Debug, PartialEq)]
+/// enum UuidVersion {
+///     V1,
+///     V2,
+///     V3,
+///     V4,
+///     V5,
+///     V6,
+///     V7,
+///     V8,
+/// }
+///
+/// // We'll use `Display` to serialize.
+/// impl fmt::Display for UuidVersion {
+///     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+///         write!(
+///             f,
+///             "{}",
+///             match self {
+///                 Self::V1 => "V1",
+///                 Self::V2 => "V2",
+///                 Self::V3 => "V3",
+///                 Self::V4 => "V4",
+///                 Self::V5 => "V5",
+///                 Self::V6 => "V6",
+///                 Self::V7 => "V7",
+///                 Self::V8 => "V8",
+///             }
+///         )
+///     }
+/// }
+///
+/// // And `FromStr` to deserialize.
+/// impl FromStr for UuidVersion {
+///     type Err = ArrowError;
+///
+///     fn from_str(s: &str) -> Result<Self, Self::Err> {
+///         match s {
+///             "V1" => Ok(Self::V1),
+///             "V2" => Ok(Self::V2),
+///             "V3" => Ok(Self::V3),
+///             "V4" => Ok(Self::V4),
+///             "V5" => Ok(Self::V5),
+///             "V6" => Ok(Self::V6),
+///             "V7" => Ok(Self::V7),
+///             "V8" => Ok(Self::V8),
+///             _ => Err(ArrowError::ParseError("Invalid UuidVersion".to_owned())),
+///         }
+///     }
+/// }
+///
+/// /// This is the extension type, not the container for Uuid values. It
+/// /// stores the Uuid version (this is the metadata of this extension type).
+/// #[derive(Clone, Copy, Debug, PartialEq)]
+/// struct Uuid(UuidVersion);
+///
+/// impl ExtensionType for Uuid {
+///     // We use a namespace as suggested by the specification.
+///     const NAME: &'static str = "myorg.example.uuid";
+///
+///     // The metadata type is the Uuid version.
+///     type Metadata = UuidVersion;
+///
+///     // We just return a reference to the Uuid version.
+///     fn metadata(&self) -> &Self::Metadata {
+///         &self.0
+///     }
+///
+///     // We use the `Display` implementation to serialize the Uuid
+///     // version.
+///     fn serialize_metadata(&self) -> Option<String> {
+///         Some(self.0.to_string())
+///     }
+///
+///     // We use the `FromStr` implementation to deserialize the Uuid
+///     // version.
+///     fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError> {
+///         metadata.map_or_else(
+///             || {
+///                 Err(ArrowError::InvalidArgumentError(
+///                     "Uuid extension type metadata missing".to_owned(),
+///                 ))
+///             },
+///             str::parse,
+///         )
+///     }
+///
+///     // The only supported data type is `FixedSizeBinary(16)`.
+///     fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError> {
+///         match data_type {
+///             DataType::FixedSizeBinary(16) => Ok(()),
+///             data_type => Err(ArrowError::InvalidArgumentError(format!(
+///                 "Uuid data type mismatch, expected FixedSizeBinary(16), found {data_type}"
+///             ))),
+///         }
+///     }
+///
+///     // We should always check if the data type is supported before
+///     // constructing the extension type.
+///     fn try_new(data_type: &DataType, metadata: Self::Metadata) -> Result<Self, ArrowError> {
+///         let uuid = Self(metadata);
+///         uuid.supports_data_type(data_type)?;
+///         Ok(uuid)
+///     }
+/// }
+///
+/// // We can now construct the extension type.
+/// let uuid_v1 = Uuid(UuidVersion::V1);
+///
+/// // And add it to a field.
+/// let mut field =
+///     Field::new("", DataType::FixedSizeBinary(16), false).with_extension_type(uuid_v1);
+///
+/// // And extract it from this field.
+/// assert_eq!(field.try_extension_type::<Uuid>()?, uuid_v1);
+///
+/// // When we try to add this to a field with an unsupported data type we
+/// // get an error.
+/// let result = Field::new("", DataType::Null, false).try_with_extension_type(uuid_v1);
+/// assert!(result.is_err());
+/// # Ok(()) }
+/// ```
+///
+/// <https://arrow.apache.org/docs/format/Columnar.html#extension-types>
+///
+/// [`Field`]: crate::Field
+pub trait ExtensionType: Sized {
+    /// The name identifying this extension type.
+    ///
+    /// This is the string value that is used for the
+    /// [`EXTENSION_TYPE_NAME_KEY`] in the [`Field::metadata`] of a [`Field`]
+    /// to identify this extension type.
+    ///
+    /// We recommend that you use a “namespace”-style prefix for extension
+    /// type names to minimize the possibility of conflicts with multiple Arrow
+    /// readers and writers in the same application. For example, use
+    /// `myorg.name_of_type` instead of simply `name_of_type`.
+    ///
+    /// Extension names beginning with `arrow.` are reserved for canonical
+    /// extension types, they should not be used for third-party extension
+    /// types.
+    ///
+    /// Extension names are case-sensitive.
+    ///
+    /// [`Field`]: crate::Field
+    /// [`Field::metadata`]: crate::Field::metadata
+    const NAME: &'static str;
+
+    /// The metadata type of this extension type.
+    ///
+    /// Implementations can use strongly or loosly typed data structures here
+    /// depending on the complexity of the metadata.
+    ///
+    /// Implementations can also use `Self` here if the extension type can be
+    /// constructed directly from its metadata.
+    ///
+    /// If an extension type defines no metadata it should use `()` to indicate
+    /// this.
+    type Metadata;
+
+    /// Returns a reference to the metadata of this extension type, or `&()` if
+    /// if this extension type defines no metadata (`Self::Metadata=()`).
+    fn metadata(&self) -> &Self::Metadata;
+
+    /// Returns the serialized representation of the metadata of this extension
+    /// type, or `None` if this extension type defines no metadata
+    /// (`Self::Metadata=()`).
+    ///
+    /// This is string value that is used for the
+    /// [`EXTENSION_TYPE_METADATA_KEY`] in the [`Field::metadata`] of a
+    /// [`Field`].
+    ///
+    /// [`Field`]: crate::Field
+    /// [`Field::metadata`]: crate::Field::metadata
+    fn serialize_metadata(&self) -> Option<String>;
+
+    /// Deserialize the metadata of this extension type from the serialized
+    /// representation of the metadata. An extension type that defines no
+    /// metadata should expect `None` for the serialized metadata and return
+    /// `Ok(())`.
+    ///
+    /// This function should return an error when
+    /// - expected metadata is missing (for extensions types with non-optional
+    ///   metadata)
+    /// - unexpected metadata is set (for extension types without metadata)
+    /// - deserialization of metadata fails
+    fn deserialize_metadata(metadata: Option<&str>) -> Result<Self::Metadata, ArrowError>;
+
+    /// Returns `OK())` iff the given data type is supported by this extension
+    /// type.
+    fn supports_data_type(&self, data_type: &DataType) -> Result<(), ArrowError>;
+
+    /// Construct this extension type for a field with the given data type and
+    /// metadata.
+    ///
+    /// This should return an error if the given data type is not supported by
+    /// this extension type.
+    fn try_new(data_type: &DataType, metadata: Self::Metadata) -> Result<Self, ArrowError>;
+}
diff --git a/arrow-schema/src/field.rs b/arrow-schema/src/field.rs
index 13bb7abf5..dbd671a62 100644
--- a/arrow-schema/src/field.rs
+++ b/arrow-schema/src/field.rs
@@ -22,8 +22,13 @@ use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
 use crate::datatype::DataType;
+#[cfg(feature = "canonical_extension_types")]
+use crate::extension::CanonicalExtensionType;
 use crate::schema::SchemaBuilder;
-use crate::{Fields, UnionFields, UnionMode};
+use crate::{
+    extension::{ExtensionType, EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY},
+    Fields, UnionFields, UnionMode,
+};
 
 /// A reference counted [`Field`]
 pub type FieldRef = Arc<Field>;
@@ -350,6 +355,167 @@ impl Field {
         self
     }
 
+    /// Returns the extension type name of this [`Field`], if set.
+    ///
+    /// This returns the value of [`EXTENSION_TYPE_NAME_KEY`], if set in
+    /// [`Field::metadata`]. If the key is missing, there is no extension type
+    /// name and this returns `None`.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use arrow_schema::{DataType, extension::EXTENSION_TYPE_NAME_KEY, Field};
+    ///
+    /// let field = Field::new("", DataType::Null, false);
+    /// assert_eq!(field.extension_type_name(), None);
+    ///
+    /// let field = Field::new("", DataType::Null, false).with_metadata(
+    ///    [(EXTENSION_TYPE_NAME_KEY.to_owned(), "example".to_owned())]
+    ///        .into_iter()
+    ///        .collect(),
+    /// );
+    /// assert_eq!(field.extension_type_name(), Some("example"));
+    /// ```
+    pub fn extension_type_name(&self) -> Option<&str> {
+        self.metadata()
+            .get(EXTENSION_TYPE_NAME_KEY)
+            .map(String::as_ref)
+    }
+
+    /// Returns the extension type metadata of this [`Field`], if set.
+    ///
+    /// This returns the value of [`EXTENSION_TYPE_METADATA_KEY`], if set in
+    /// [`Field::metadata`]. If the key is missing, there is no extension type
+    /// metadata and this returns `None`.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use arrow_schema::{DataType, extension::EXTENSION_TYPE_METADATA_KEY, Field};
+    ///
+    /// let field = Field::new("", DataType::Null, false);
+    /// assert_eq!(field.extension_type_metadata(), None);
+    ///
+    /// let field = Field::new("", DataType::Null, false).with_metadata(
+    ///    [(EXTENSION_TYPE_METADATA_KEY.to_owned(), "example".to_owned())]
+    ///        .into_iter()
+    ///        .collect(),
+    /// );
+    /// assert_eq!(field.extension_type_metadata(), Some("example"));
+    /// ```
+    pub fn extension_type_metadata(&self) -> Option<&str> {
+        self.metadata()
+            .get(EXTENSION_TYPE_METADATA_KEY)
+            .map(String::as_ref)
+    }
+
+    /// Returns an instance of the given [`ExtensionType`] of this [`Field`],
+    /// if set in the [`Field::metadata`].
+    ///
+    /// # Error
+    ///
+    /// Returns an error if
+    /// - this field does not have the name of this extension type
+    ///   ([`ExtensionType::NAME`]) in the [`Field::metadata`] (mismatch or
+    ///   missing)
+    /// - the deserialization of the metadata
+    ///   ([`ExtensionType::deserialize_metadata`]) fails
+    /// - the construction of the extension type ([`ExtensionType::try_new`])
+    ///   fail (for example when the [`Field::data_type`] is not supported by
+    ///   the extension type ([`ExtensionType::supports_data_type`]))
+    pub fn try_extension_type<E: ExtensionType>(&self) -> Result<E, ArrowError> {
+        // Check the extension name in the metadata
+        match self.extension_type_name() {
+            // It should match the name of the given extension type
+            Some(name) if name == E::NAME => {
+                // Deserialize the metadata and try to construct the extension
+                // type
+                E::deserialize_metadata(self.extension_type_metadata())
+                    .and_then(|metadata| E::try_new(self.data_type(), metadata))
+            }
+            // Name mismatch
+            Some(name) => Err(ArrowError::InvalidArgumentError(format!(
+                "Field extension type name mismatch, expected {}, found {name}",
+                E::NAME
+            ))),
+            // Name missing
+            None => Err(ArrowError::InvalidArgumentError(
+                "Field extension type name missing".to_owned(),
+            )),
+        }
+    }
+
+    /// Returns an instance of the given [`ExtensionType`] of this [`Field`],
+    /// panics if this [`Field`] does not have this extension type.
+    ///
+    /// # Panic
+    ///
+    /// This calls [`Field::try_extension_type`] and panics when it returns an
+    /// error.
+    pub fn extension_type<E: ExtensionType>(&self) -> E {
+        self.try_extension_type::<E>()
+            .unwrap_or_else(|e| panic!("{e}"))
+    }
+
+    /// Updates the metadata of this [`Field`] with the [`ExtensionType::NAME`]
+    /// and [`ExtensionType::metadata`] of the given [`ExtensionType`], if the
+    /// given extension type supports the [`Field::data_type`] of this field
+    /// ([`ExtensionType::supports_data_type`]).
+    ///
+    /// If the given extension type defines no metadata, a previously set
+    /// value of [`EXTENSION_TYPE_METADATA_KEY`] is cleared.
+    ///
+    /// # Error
+    ///
+    /// This functions returns an error if the data type of this field does not
+    /// match any of the supported storage types of the given extension type.
+    pub fn try_with_extension_type<E: ExtensionType>(
+        &mut self,
+        extension_type: E,
+    ) -> Result<(), ArrowError> {
+        // Make sure the data type of this field is supported
+        extension_type.supports_data_type(&self.data_type)?;
+
+        self.metadata
+            .insert(EXTENSION_TYPE_NAME_KEY.to_owned(), E::NAME.to_owned());
+        match extension_type.serialize_metadata() {
+            Some(metadata) => self
+                .metadata
+                .insert(EXTENSION_TYPE_METADATA_KEY.to_owned(), metadata),
+            // If this extension type has no metadata, we make sure to
+            // clear previously set metadata.
+            None => self.metadata.remove(EXTENSION_TYPE_METADATA_KEY),
+        };
+
+        Ok(())
+    }
+
+    /// Updates the metadata of this [`Field`] with the [`ExtensionType::NAME`]
+    /// and [`ExtensionType::metadata`] of the given [`ExtensionType`].
+    ///
+    /// # Panics
+    ///
+    /// This calls [`Field::try_with_extension_type`] and panics when it
+    /// returns an error.
+    pub fn with_extension_type<E: ExtensionType>(mut self, extension_type: E) -> Self {
+        self.try_with_extension_type(extension_type)
+            .unwrap_or_else(|e| panic!("{e}"));
+        self
+    }
+
+    /// Returns the [`CanonicalExtensionType`] of this [`Field`], if set.
+    ///
+    /// # Error
+    ///
+    /// Returns an error if
+    /// - this field does have a canonical extension type (mismatch or missing)
+    /// - the canonical extension is not supported
+    /// - the construction of the extension type fails
+    #[cfg(feature = "canonical_extension_types")]
+    pub fn try_canonical_extension_type(&self) -> Result<CanonicalExtensionType, ArrowError> {
+        CanonicalExtensionType::try_from(self)
+    }
+
     /// Indicates whether this [`Field`] supports null values.
     #[inline]
     pub const fn is_nullable(&self) -> bool {
diff --git a/arrow-schema/src/lib.rs b/arrow-schema/src/lib.rs
index d06382fbc..a83e23e27 100644
--- a/arrow-schema/src/lib.rs
+++ b/arrow-schema/src/lib.rs
@@ -25,6 +25,7 @@ use std::fmt::Display;
 mod datatype_parse;
 mod error;
 pub use error::*;
+pub mod extension;
 mod field;
 pub use field::*;
 mod fields;
diff --git a/arrow-select/Cargo.toml b/arrow-select/Cargo.toml
index 023788799..0d412d151 100644
--- a/arrow-select/Cargo.toml
+++ b/arrow-select/Cargo.toml
@@ -45,4 +45,4 @@ ahash = { version = "0.8", default-features = false}
 default = []
 
 [dev-dependencies]
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
diff --git a/arrow-select/src/dictionary.rs b/arrow-select/src/dictionary.rs
index 2a532600b..c363b9992 100644
--- a/arrow-select/src/dictionary.rs
+++ b/arrow-select/src/dictionary.rs
@@ -315,7 +315,7 @@ mod tests {
         assert_eq!(merged.values.as_ref(), &expected);
         assert_eq!(merged.key_mappings.len(), 2);
         assert_eq!(&merged.key_mappings[0], &[0, 0, 0, 1, 0]);
-        assert_eq!(&merged.key_mappings[1], &[]);
+        assert_eq!(&merged.key_mappings[1], &[] as &[i32; 0]);
     }
 
     #[test]
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index c91732848..7bb140d37 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -864,8 +864,10 @@ mod tests {
     use arrow_array::builder::*;
     use arrow_array::cast::as_run_array;
     use arrow_array::types::*;
-    use rand::distributions::{Alphanumeric, Standard};
+    use rand::distr::uniform::{UniformSampler, UniformUsize};
+    use rand::distr::{Alphanumeric, StandardUniform};
     use rand::prelude::*;
+    use rand::rng;
 
     use super::*;
 
@@ -1475,9 +1477,9 @@ mod tests {
     }
 
     fn test_slices_fuzz(mask_len: usize, offset: usize, truncate: usize) {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
-        let bools: Vec<bool> = std::iter::from_fn(|| Some(rng.gen()))
+        let bools: Vec<bool> = std::iter::from_fn(|| Some(rng.random()))
             .take(mask_len)
             .collect();
 
@@ -1516,15 +1518,19 @@ mod tests {
     #[test]
     #[cfg_attr(miri, ignore)]
     fn fuzz_test_slices_iterator() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
+        let uusize = UniformUsize::new(usize::MIN, usize::MAX).unwrap();
         for _ in 0..100 {
-            let mask_len = rng.gen_range(0..1024);
+            let mask_len = rng.random_range(0..1024);
             let max_offset = 64.min(mask_len);
-            let offset = rng.gen::<usize>().checked_rem(max_offset).unwrap_or(0);
+            let offset = uusize.sample(&mut rng).checked_rem(max_offset).unwrap_or(0);
 
             let max_truncate = 128.min(mask_len - offset);
-            let truncate = rng.gen::<usize>().checked_rem(max_truncate).unwrap_or(0);
+            let truncate = uusize
+                .sample(&mut rng)
+                .checked_rem(max_truncate)
+                .unwrap_or(0);
 
             test_slices_fuzz(mask_len, offset, truncate);
         }
@@ -1549,11 +1555,11 @@ mod tests {
     /// Generates an array of length `len` with `valid_percent` non-null values
     fn gen_primitive<T>(len: usize, valid_percent: f64) -> Vec<Option<T>>
     where
-        Standard: Distribution<T>,
+        StandardUniform: Distribution<T>,
     {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         (0..len)
-            .map(|_| rng.gen_bool(valid_percent).then(|| rng.gen()))
+            .map(|_| rng.random_bool(valid_percent).then(|| rng.random()))
             .collect()
     }
 
@@ -1563,11 +1569,11 @@ mod tests {
         valid_percent: f64,
         str_len_range: std::ops::Range<usize>,
     ) -> Vec<Option<String>> {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         (0..len)
             .map(|_| {
-                rng.gen_bool(valid_percent).then(|| {
-                    let len = rng.gen_range(str_len_range.clone());
+                rng.random_bool(valid_percent).then(|| {
+                    let len = rng.random_range(str_len_range.clone());
                     (0..len)
                         .map(|_| char::from(rng.sample(Alphanumeric)))
                         .collect()
@@ -1584,24 +1590,24 @@ mod tests {
     #[test]
     #[cfg_attr(miri, ignore)]
     fn fuzz_filter() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
         for i in 0..100 {
             let filter_percent = match i {
                 0..=4 => 1.,
                 5..=10 => 0.,
-                _ => rng.gen_range(0.0..1.0),
+                _ => rng.random_range(0.0..1.0),
             };
 
-            let valid_percent = rng.gen_range(0.0..1.0);
+            let valid_percent = rng.random_range(0.0..1.0);
 
-            let array_len = rng.gen_range(32..256);
-            let array_offset = rng.gen_range(0..10);
+            let array_len = rng.random_range(32..256);
+            let array_offset = rng.random_range(0..10);
 
             // Construct a predicate
-            let filter_offset = rng.gen_range(0..10);
-            let filter_truncate = rng.gen_range(0..10);
-            let bools: Vec<_> = std::iter::from_fn(|| Some(rng.gen_bool(filter_percent)))
+            let filter_offset = rng.random_range(0..10);
+            let filter_truncate = rng.random_range(0..10);
+            let bools: Vec<_> = std::iter::from_fn(|| Some(rng.random_bool(filter_percent)))
                 .take(array_len + filter_offset - filter_truncate)
                 .collect();
 
diff --git a/arrow-select/src/nullif.rs b/arrow-select/src/nullif.rs
index 4b90114a4..dc729da7e 100644
--- a/arrow-select/src/nullif.rs
+++ b/arrow-select/src/nullif.rs
@@ -120,7 +120,7 @@ mod tests {
     use arrow_array::{Int32Array, NullArray, StringArray, StructArray};
     use arrow_data::ArrayData;
     use arrow_schema::{Field, Fields};
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
 
     #[test]
     fn test_nullif_int_array() {
@@ -497,11 +497,13 @@ mod tests {
 
     #[test]
     fn nullif_fuzz() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
 
         let arrays = [
             Int32Array::from(vec![0; 128]),
-            (0..128).map(|_| rng.gen_bool(0.5).then_some(0)).collect(),
+            (0..128)
+                .map(|_| rng.random_bool(0.5).then_some(0))
+                .collect(),
         ];
 
         for a in arrays {
@@ -511,11 +513,11 @@ mod tests {
                 let a = a.slice(a_offset, a_length);
 
                 for i in 1..65 {
-                    let b_start_offset = rng.gen_range(0..i);
-                    let b_end_offset = rng.gen_range(0..i);
+                    let b_start_offset = rng.random_range(0..i);
+                    let b_end_offset = rng.random_range(0..i);
 
                     let b: BooleanArray = (0..a_length + b_start_offset + b_end_offset)
-                        .map(|_| rng.gen_bool(0.5).then(|| rng.gen_bool(0.5)))
+                        .map(|_| rng.random_bool(0.5).then(|| rng.random_bool(0.5)))
                         .collect();
                     let b = b.slice(b_start_offset, a_length);
 
diff --git a/arrow/Cargo.toml b/arrow/Cargo.toml
index 1b01dcd25..f621e9d3b 100644
--- a/arrow/Cargo.toml
+++ b/arrow/Cargo.toml
@@ -53,7 +53,7 @@ arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
 arrow-string = { workspace = true }
 
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"], optional = true }
 pyo3 = { version = "0.23", default-features = false, optional = true }
 half = { version = "2.1", default-features = false, optional = true }
 
@@ -80,12 +80,13 @@ force_validate = ["arrow-array/force_validate", "arrow-data/force_validate"]
 # Enable ffi support
 ffi = ["arrow-schema/ffi", "arrow-data/ffi", "arrow-array/ffi"]
 chrono-tz = ["arrow-array/chrono-tz"]
+canonical_extension_types = ["arrow-schema/canonical_extension_types"]
 
 [dev-dependencies]
 chrono = { workspace = true }
 criterion = { version = "0.5", default-features = false }
 half = { version = "2.1", default-features = false }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 serde = { version = "1.0", default-features = false, features = ["derive"] }
 # used in examples
 memmap2 = "0.9.3"
diff --git a/arrow/README.md b/arrow/README.md
index 79aefaae9..64d9eb980 100644
--- a/arrow/README.md
+++ b/arrow/README.md
@@ -61,6 +61,7 @@ The `arrow` crate provides the following features which may be enabled in your `
 - `chrono-tz` - support of parsing timezone using [chrono-tz](https://docs.rs/chrono-tz/0.6.0/chrono_tz/)
 - `ffi` - bindings for the Arrow C [C Data Interface](https://arrow.apache.org/docs/format/CDataInterface.html)
 - `pyarrow` - bindings for pyo3 to call arrow-rs from python
+- `canonical_extension_types` - definitions for [canonical extension types](https://arrow.apache.org/docs/format/CanonicalExtensions.html#format-canonical-extensions)
 
 ## Arrow Feature Status
 
diff --git a/arrow/benches/aggregate_kernels.rs b/arrow/benches/aggregate_kernels.rs
index 6e224a48c..25dbe3548 100644
--- a/arrow/benches/aggregate_kernels.rs
+++ b/arrow/benches/aggregate_kernels.rs
@@ -18,7 +18,7 @@
 #[macro_use]
 extern crate criterion;
 use criterion::{Criterion, Throughput};
-use rand::distributions::{Distribution, Standard};
+use rand::distr::{Distribution, StandardUniform};
 
 extern crate arrow;
 
@@ -31,7 +31,7 @@ const BATCH_SIZE: usize = 64 * 1024;
 
 fn primitive_benchmark<T: ArrowNumericType>(c: &mut Criterion, name: &str)
 where
-    Standard: Distribution<T::Native>,
+    StandardUniform: Distribution<T::Native>,
 {
     let nonnull_array = create_primitive_array::<T>(BATCH_SIZE, 0.0);
     let nullable_array = create_primitive_array::<T>(BATCH_SIZE, 0.5);
diff --git a/arrow/benches/array_from_vec.rs b/arrow/benches/array_from_vec.rs
index fd83ad5c2..c256d1523 100644
--- a/arrow/benches/array_from_vec.rs
+++ b/arrow/benches/array_from_vec.rs
@@ -99,10 +99,10 @@ fn decimal_benchmark(c: &mut Criterion) {
     // bench decimal128 array
     // create option<i128> array
     let size: usize = 1 << 15;
-    let mut rng = rand::thread_rng();
+    let mut rng = rand::rng();
     let mut array = vec![];
     for _ in 0..size {
-        array.push(Some(rng.gen_range::<i128, _>(0..9999999999)));
+        array.push(Some(rng.random_range::<i128, _>(0..9999999999)));
     }
     c.bench_function("decimal128_array_from_vec 32768", |b| {
         b.iter(|| decimal128_array_from_vec(array.as_slice()))
@@ -112,9 +112,9 @@ fn decimal_benchmark(c: &mut Criterion) {
     // create option<into<decimal256>> array
     let size = 1 << 10;
     let mut array = vec![];
-    let mut rng = rand::thread_rng();
+    let mut rng = rand::rng();
     for _ in 0..size {
-        let decimal = i256::from_i128(rng.gen_range::<i128, _>(0..9999999999999));
+        let decimal = i256::from_i128(rng.random_range::<i128, _>(0..9999999999999));
         array.push(Some(decimal));
     }
 
diff --git a/arrow/benches/boolean_append_packed.rs b/arrow/benches/boolean_append_packed.rs
index 40873422d..508720eb3 100644
--- a/arrow/benches/boolean_append_packed.rs
+++ b/arrow/benches/boolean_append_packed.rs
@@ -17,22 +17,22 @@
 
 use arrow::array::BooleanBufferBuilder;
 use criterion::{criterion_group, criterion_main, Criterion};
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 
 fn rand_bytes(len: usize) -> Vec<u8> {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let mut buf = vec![0_u8; len];
     rng.fill(buf.as_mut_slice());
     buf
 }
 
 fn boolean_append_packed(c: &mut Criterion) {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let source = rand_bytes(1024);
     let ranges: Vec<_> = (0..100)
         .map(|_| {
-            let start: usize = rng.gen_range(0..1024 * 8);
-            let end: usize = rng.gen_range(start..1024 * 8);
+            let start: usize = rng.random_range(0..1024 * 8);
+            let end: usize = rng.random_range(start..1024 * 8);
             start..end
         })
         .collect();
diff --git a/arrow/benches/buffer_create.rs b/arrow/benches/buffer_create.rs
index e7d24c216..f6199ccab 100644
--- a/arrow/benches/buffer_create.rs
+++ b/arrow/benches/buffer_create.rs
@@ -19,7 +19,7 @@
 extern crate criterion;
 use arrow::util::test_util::seedable_rng;
 use criterion::Criterion;
-use rand::distributions::Uniform;
+use rand::distr::Uniform;
 use rand::Rng;
 
 extern crate arrow;
@@ -110,7 +110,7 @@ fn from_slice(data: &[Vec<u32>], capacity: usize) -> Buffer {
 
 fn create_data(size: usize) -> Vec<Vec<u32>> {
     let rng = &mut seedable_rng();
-    let range = Uniform::new(0, 33);
+    let range = Uniform::new(0, 33).unwrap();
 
     (0..size)
         .map(|_| {
@@ -125,7 +125,7 @@ fn create_data(size: usize) -> Vec<Vec<u32>> {
 
 fn create_data_bool(size: usize) -> Vec<Vec<bool>> {
     let rng = &mut seedable_rng();
-    let range = Uniform::new(0, 33);
+    let range = Uniform::new(0, 33).unwrap();
 
     (0..size)
         .map(|_| {
diff --git a/arrow/benches/builder.rs b/arrow/benches/builder.rs
index 87a02e7ad..2776924d8 100644
--- a/arrow/benches/builder.rs
+++ b/arrow/benches/builder.rs
@@ -22,7 +22,7 @@ extern crate rand;
 use std::mem::size_of;
 
 use criterion::*;
-use rand::distributions::Standard;
+use rand::distr::StandardUniform;
 
 use arrow::array::*;
 use arrow::util::test_util::seedable_rng;
@@ -68,7 +68,7 @@ fn bench_primitive_nulls(c: &mut Criterion) {
 
 fn bench_bool(c: &mut Criterion) {
     let data: Vec<bool> = seedable_rng()
-        .sample_iter(&Standard)
+        .sample_iter(&StandardUniform)
         .take(BATCH_SIZE)
         .collect();
     let data_len = data.len();
@@ -110,10 +110,10 @@ fn bench_string(c: &mut Criterion) {
 fn bench_decimal128(c: &mut Criterion) {
     c.bench_function("bench_decimal128_builder", |b| {
         b.iter(|| {
-            let mut rng = rand::thread_rng();
+            let mut rng = rand::rng();
             let mut decimal_builder = Decimal128Builder::with_capacity(BATCH_SIZE);
             for _ in 0..BATCH_SIZE {
-                decimal_builder.append_value(rng.gen_range::<i128, _>(0..9999999999));
+                decimal_builder.append_value(rng.random_range::<i128, _>(0..9999999999));
             }
             black_box(
                 decimal_builder
@@ -128,11 +128,11 @@ fn bench_decimal128(c: &mut Criterion) {
 fn bench_decimal256(c: &mut Criterion) {
     c.bench_function("bench_decimal128_builder", |b| {
         b.iter(|| {
-            let mut rng = rand::thread_rng();
+            let mut rng = rand::rng();
             let mut decimal_builder = Decimal256Builder::with_capacity(BATCH_SIZE);
             for _ in 0..BATCH_SIZE {
                 decimal_builder
-                    .append_value(i256::from_i128(rng.gen_range::<i128, _>(0..99999999999)));
+                    .append_value(i256::from_i128(rng.random_range::<i128, _>(0..99999999999)));
             }
             black_box(
                 decimal_builder
diff --git a/arrow/benches/cast_kernels.rs b/arrow/benches/cast_kernels.rs
index 5c4fcff13..9c3063f20 100644
--- a/arrow/benches/cast_kernels.rs
+++ b/arrow/benches/cast_kernels.rs
@@ -18,7 +18,7 @@
 #[macro_use]
 extern crate criterion;
 use criterion::Criterion;
-use rand::distributions::{Distribution, Standard, Uniform};
+use rand::distr::{Distribution, StandardUniform, Uniform};
 use rand::Rng;
 
 use chrono::DateTime;
@@ -34,7 +34,7 @@ use arrow::util::test_util::seedable_rng;
 
 fn build_array<T: ArrowPrimitiveType>(size: usize) -> ArrayRef
 where
-    Standard: Distribution<T::Native>,
+    StandardUniform: Distribution<T::Native>,
 {
     let array = create_primitive_array::<T>(size, 0.1);
     Arc::new(array)
@@ -46,10 +46,10 @@ fn build_utf8_date_array(size: usize, with_nulls: bool) -> ArrayRef {
     // use random numbers to avoid spurious compiler optimizations wrt to branching
     let mut rng = seedable_rng();
     let mut builder = StringBuilder::new();
-    let range = Uniform::new(0, 737776);
+    let range = Uniform::new(0, 737776).unwrap();
 
     for _ in 0..size {
-        if with_nulls && rng.gen::<f32>() > 0.8 {
+        if with_nulls && rng.random::<f32>() > 0.8 {
             builder.append_null();
         } else {
             let string = NaiveDate::from_num_days_from_ce_opt(rng.sample(range))
@@ -66,10 +66,10 @@ fn build_utf8_date_time_array(size: usize, with_nulls: bool) -> ArrayRef {
     // use random numbers to avoid spurious compiler optimizations wrt to branching
     let mut rng = seedable_rng();
     let mut builder = StringBuilder::new();
-    let range = Uniform::new(0, 1608071414123);
+    let range = Uniform::new(0, 1608071414123).unwrap();
 
     for _ in 0..size {
-        if with_nulls && rng.gen::<f32>() > 0.8 {
+        if with_nulls && rng.random::<f32>() > 0.8 {
             builder.append_null();
         } else {
             let string = DateTime::from_timestamp(rng.sample(range), 0)
@@ -87,7 +87,7 @@ fn build_decimal128_array(size: usize, precision: u8, scale: i8) -> ArrayRef {
     let mut builder = Decimal128Builder::with_capacity(size);
 
     for _ in 0..size {
-        builder.append_value(rng.gen_range::<i128, _>(0..1000000000));
+        builder.append_value(rng.random_range::<i128, _>(0..1000000000));
     }
     Arc::new(
         builder
@@ -102,7 +102,7 @@ fn build_decimal256_array(size: usize, precision: u8, scale: i8) -> ArrayRef {
     let mut builder = Decimal256Builder::with_capacity(size);
     let mut bytes = [0; 32];
     for _ in 0..size {
-        let num = rng.gen_range::<i128, _>(0..1000000000);
+        let num = rng.random_range::<i128, _>(0..1000000000);
         bytes[0..16].clone_from_slice(&num.to_le_bytes());
         builder.append_value(i256::from_le_bytes(bytes));
     }
diff --git a/arrow/benches/comparison_kernels.rs b/arrow/benches/comparison_kernels.rs
index 4c4a63a77..84fd47acc 100644
--- a/arrow/benches/comparison_kernels.rs
+++ b/arrow/benches/comparison_kernels.rs
@@ -72,8 +72,8 @@ fn bench_string_regexp_is_match_scalar(arr_a: &StringArray, value_b: &str) {
 
 fn make_string_array(size: usize, rng: &mut StdRng) -> impl Iterator<Item = Option<String>> + '_ {
     (0..size).map(|_| {
-        let len = rng.gen_range(0..64);
-        let bytes = (0..len).map(|_| rng.gen_range(0..128)).collect();
+        let len = rng.random_range(0..64);
+        let bytes = (0..len).map(|_| rng.random_range(0..128)).collect();
         Some(String::from_utf8(bytes).unwrap())
     })
 }
diff --git a/arrow/benches/csv_reader.rs b/arrow/benches/csv_reader.rs
index 74a47ef89..331ff9edd 100644
--- a/arrow/benches/csv_reader.rs
+++ b/arrow/benches/csv_reader.rs
@@ -61,45 +61,45 @@ fn criterion_benchmark(c: &mut Criterion) {
     let mut rng = seedable_rng();
 
     // Single Primitive Column tests
-    let values = Int32Array::from_iter_values((0..4096).map(|_| rng.gen_range(0..1024)));
+    let values = Int32Array::from_iter_values((0..4096).map(|_| rng.random_range(0..1024)));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 i32_small(0)", cols);
 
-    let values = Int32Array::from_iter_values((0..4096).map(|_| rng.gen()));
+    let values = Int32Array::from_iter_values((0..4096).map(|_| rng.random()));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 i32(0)", cols);
 
-    let values = UInt64Array::from_iter_values((0..4096).map(|_| rng.gen_range(0..1024)));
+    let values = UInt64Array::from_iter_values((0..4096).map(|_| rng.random_range(0..1024)));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 u64_small(0)", cols);
 
-    let values = UInt64Array::from_iter_values((0..4096).map(|_| rng.gen()));
+    let values = UInt64Array::from_iter_values((0..4096).map(|_| rng.random()));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 u64(0)", cols);
 
-    let values = Int64Array::from_iter_values((0..4096).map(|_| rng.gen_range(0..1024) - 512));
+    let values = Int64Array::from_iter_values((0..4096).map(|_| rng.random_range(0..1024) - 512));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 i64_small(0)", cols);
 
-    let values = Int64Array::from_iter_values((0..4096).map(|_| rng.gen()));
+    let values = Int64Array::from_iter_values((0..4096).map(|_| rng.random()));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 i64(0)", cols);
 
     let cols = vec![Arc::new(Float32Array::from_iter_values(
-        (0..4096).map(|_| rng.gen_range(0..1024000) as f32 / 1000.),
+        (0..4096).map(|_| rng.random_range(0..1024000) as f32 / 1000.),
     )) as _];
     do_bench(c, "4096 f32_small(0)", cols);
 
-    let values = Float32Array::from_iter_values((0..4096).map(|_| rng.gen()));
+    let values = Float32Array::from_iter_values((0..4096).map(|_| rng.random()));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 f32(0)", cols);
 
     let cols = vec![Arc::new(Float64Array::from_iter_values(
-        (0..4096).map(|_| rng.gen_range(0..1024000) as f64 / 1000.),
+        (0..4096).map(|_| rng.random_range(0..1024000) as f64 / 1000.),
     )) as _];
     do_bench(c, "4096 f64_small(0)", cols);
 
-    let values = Float64Array::from_iter_values((0..4096).map(|_| rng.gen()));
+    let values = Float64Array::from_iter_values((0..4096).map(|_| rng.random()));
     let cols = vec![Arc::new(values) as ArrayRef];
     do_bench(c, "4096 f64(0)", cols);
 
diff --git a/arrow/benches/decimal_validate.rs b/arrow/benches/decimal_validate.rs
index be812a225..dfa4f5992 100644
--- a/arrow/benches/decimal_validate.rs
+++ b/arrow/benches/decimal_validate.rs
@@ -35,11 +35,11 @@ fn validate_decimal256_array(array: Decimal256Array) {
 }
 
 fn validate_decimal128_benchmark(c: &mut Criterion) {
-    let mut rng = rand::thread_rng();
+    let mut rng = rand::rng();
     let size: i128 = 20000;
     let mut decimal_builder = Decimal128Builder::with_capacity(size as usize);
     for _ in 0..size {
-        decimal_builder.append_value(rng.gen_range::<i128, _>(0..999999999999));
+        decimal_builder.append_value(rng.random_range::<i128, _>(0..999999999999));
     }
     let decimal_array = decimal_builder
         .finish()
@@ -55,11 +55,11 @@ fn validate_decimal128_benchmark(c: &mut Criterion) {
 }
 
 fn validate_decimal256_benchmark(c: &mut Criterion) {
-    let mut rng = rand::thread_rng();
+    let mut rng = rand::rng();
     let size: i128 = 20000;
     let mut decimal_builder = Decimal256Builder::with_capacity(size as usize);
     for _ in 0..size {
-        let v = rng.gen_range::<i128, _>(0..999999999999999);
+        let v = rng.random_range::<i128, _>(0..999999999999999);
         let decimal = i256::from_i128(v);
         decimal_builder.append_value(decimal);
     }
diff --git a/arrow/benches/interleave_kernels.rs b/arrow/benches/interleave_kernels.rs
index 0941f1e3f..ed7ac1237 100644
--- a/arrow/benches/interleave_kernels.rs
+++ b/arrow/benches/interleave_kernels.rs
@@ -54,8 +54,8 @@ fn bench_values(c: &mut Criterion, name: &str, len: usize, values: &[&dyn Array]
     let mut rng = seedable_rng();
     let indices: Vec<_> = (0..len)
         .map(|_| {
-            let array_idx = rng.gen_range(0..values.len());
-            let value_idx = rng.gen_range(0..values[array_idx].len());
+            let array_idx = rng.random_range(0..values.len());
+            let value_idx = rng.random_range(0..values[array_idx].len());
             (array_idx, value_idx)
         })
         .collect();
diff --git a/arrow/benches/json_writer.rs b/arrow/benches/json_writer.rs
index 48be0bccb..ff76ecdd6 100644
--- a/arrow/benches/json_writer.rs
+++ b/arrow/benches/json_writer.rs
@@ -61,7 +61,7 @@ fn create_mixed(len: usize) -> RecordBatch {
 
 fn create_nulls(len: usize) -> NullBuffer {
     let mut rng = seedable_rng();
-    BooleanBuffer::from_iter((0..len).map(|_| rng.gen_bool(0.2))).into()
+    BooleanBuffer::from_iter((0..len).map(|_| rng.random_bool(0.2))).into()
 }
 
 fn create_offsets(len: usize) -> (usize, OffsetBuffer<i32>) {
@@ -70,7 +70,7 @@ fn create_offsets(len: usize) -> (usize, OffsetBuffer<i32>) {
     let mut offsets = Vec::with_capacity(len + 1);
     offsets.push(0);
     for _ in 0..len {
-        let len = rng.gen_range(0..10);
+        let len = rng.random_range(0..10);
         offsets.push(last_offset + len);
         last_offset += len;
     }
diff --git a/arrow/benches/mutable_array.rs b/arrow/benches/mutable_array.rs
index b04e5cd84..67591194a 100644
--- a/arrow/benches/mutable_array.rs
+++ b/arrow/benches/mutable_array.rs
@@ -31,8 +31,8 @@ fn create_slices(size: usize) -> Vec<(usize, usize)> {
 
     (0..size)
         .map(|_| {
-            let start = rng.gen_range(0..size / 2);
-            let end = rng.gen_range(start + 1..size);
+            let start = rng.random_range(0..size / 2);
+            let end = rng.random_range(start + 1..size);
             (start, end)
         })
         .collect()
diff --git a/arrow/benches/partition_kernels.rs b/arrow/benches/partition_kernels.rs
index fce8634a1..e6a067def 100644
--- a/arrow/benches/partition_kernels.rs
+++ b/arrow/benches/partition_kernels.rs
@@ -27,12 +27,12 @@ use arrow::{
     datatypes::{Float64Type, UInt8Type},
 };
 use arrow_ord::partition::partition;
-use rand::distributions::{Distribution, Standard};
+use rand::distr::{Distribution, StandardUniform};
 use std::iter;
 
 fn create_array<T: ArrowPrimitiveType>(size: usize, with_nulls: bool) -> ArrayRef
 where
-    Standard: Distribution<T::Native>,
+    StandardUniform: Distribution<T::Native>,
 {
     let null_density = if with_nulls { 0.5 } else { 0.0 };
     let array = create_primitive_array::<T>(size, null_density);
diff --git a/arrow/benches/primitive_run_take.rs b/arrow/benches/primitive_run_take.rs
index c10c16bfe..cabf9c118 100644
--- a/arrow/benches/primitive_run_take.rs
+++ b/arrow/benches/primitive_run_take.rs
@@ -28,10 +28,10 @@ fn create_random_index(size: usize, null_density: f32, max_value: usize) -> UInt
     let mut rng = seedable_rng();
     let mut builder = UInt32Builder::with_capacity(size);
     for _ in 0..size {
-        if rng.gen::<f32>() < null_density {
+        if rng.random::<f32>() < null_density {
             builder.append_null();
         } else {
-            let value = rng.gen_range::<u32, _>(0u32..max_value as u32);
+            let value = rng.random_range::<u32, _>(0u32..max_value as u32);
             builder.append_value(value);
         }
     }
diff --git a/arrow/benches/string_dictionary_builder.rs b/arrow/benches/string_dictionary_builder.rs
index 424400674..a39fd5d03 100644
--- a/arrow/benches/string_dictionary_builder.rs
+++ b/arrow/benches/string_dictionary_builder.rs
@@ -18,17 +18,17 @@
 use arrow::array::StringDictionaryBuilder;
 use arrow::datatypes::Int32Type;
 use criterion::{criterion_group, criterion_main, Criterion};
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 
 /// Note: this is best effort, not all keys are necessarily present or unique
 fn build_strings(dict_size: usize, total_size: usize, key_len: usize) -> Vec<String> {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let values: Vec<String> = (0..dict_size)
-        .map(|_| (0..key_len).map(|_| rng.gen::<char>()).collect())
+        .map(|_| (0..key_len).map(|_| rng.random::<char>()).collect())
         .collect();
 
     (0..total_size)
-        .map(|_| values[rng.gen_range(0..dict_size)].clone())
+        .map(|_| values[rng.random_range(0..dict_size)].clone())
         .collect()
 }
 
diff --git a/arrow/benches/string_run_iterator.rs b/arrow/benches/string_run_iterator.rs
index ac5cf7838..32088573d 100644
--- a/arrow/benches/string_run_iterator.rs
+++ b/arrow/benches/string_run_iterator.rs
@@ -18,17 +18,17 @@
 use arrow::array::{Int32RunArray, StringArray, StringRunBuilder};
 use arrow::datatypes::Int32Type;
 use criterion::{criterion_group, criterion_main, Criterion};
-use rand::{thread_rng, Rng};
+use rand::{rng, Rng};
 
 fn build_strings_runs(
     physical_array_len: usize,
     logical_array_len: usize,
     string_len: usize,
 ) -> Int32RunArray {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     let run_len = logical_array_len / physical_array_len;
     let mut values: Vec<String> = (0..physical_array_len)
-        .map(|_| (0..string_len).map(|_| rng.gen::<char>()).collect())
+        .map(|_| (0..string_len).map(|_| rng.random::<char>()).collect())
         .flat_map(|s| std::iter::repeat(s).take(run_len))
         .collect();
     while values.len() < logical_array_len {
diff --git a/arrow/benches/take_kernels.rs b/arrow/benches/take_kernels.rs
index 77ec54c97..a09064839 100644
--- a/arrow/benches/take_kernels.rs
+++ b/arrow/benches/take_kernels.rs
@@ -32,10 +32,10 @@ fn create_random_index(size: usize, null_density: f32) -> UInt32Array {
     let mut rng = seedable_rng();
     let mut builder = UInt32Builder::with_capacity(size);
     for _ in 0..size {
-        if rng.gen::<f32>() < null_density {
+        if rng.random::<f32>() < null_density {
             builder.append_null();
         } else {
-            let value = rng.gen_range::<u32, _>(0u32..size as u32);
+            let value = rng.random_range::<u32, _>(0u32..size as u32);
             builder.append_value(value);
         }
     }
diff --git a/arrow/src/util/bench_util.rs b/arrow/src/util/bench_util.rs
index 53e010341..387d9b973 100644
--- a/arrow/src/util/bench_util.rs
+++ b/arrow/src/util/bench_util.rs
@@ -22,12 +22,12 @@ use crate::datatypes::*;
 use crate::util::test_util::seedable_rng;
 use arrow_buffer::{Buffer, IntervalMonthDayNano};
 use half::f16;
-use rand::distributions::uniform::SampleUniform;
-use rand::thread_rng;
+use rand::distr::uniform::SampleUniform;
+use rand::rng;
 use rand::Rng;
 use rand::SeedableRng;
 use rand::{
-    distributions::{Alphanumeric, Distribution, Standard},
+    distr::{Alphanumeric, Distribution, StandardUniform},
     prelude::StdRng,
 };
 use std::ops::Range;
@@ -36,16 +36,16 @@ use std::ops::Range;
 pub fn create_primitive_array<T>(size: usize, null_density: f32) -> PrimitiveArray<T>
 where
     T: ArrowPrimitiveType,
-    Standard: Distribution<T::Native>,
+    StandardUniform: Distribution<T::Native>,
 {
     let mut rng = seedable_rng();
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
-                Some(rng.gen())
+                Some(rng.random())
             }
         })
         .collect()
@@ -60,16 +60,16 @@ pub fn create_primitive_array_with_seed<T>(
 ) -> PrimitiveArray<T>
 where
     T: ArrowPrimitiveType,
-    Standard: Distribution<T::Native>,
+    StandardUniform: Distribution<T::Native>,
 {
     let mut rng = StdRng::seed_from_u64(seed);
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
-                Some(rng.gen())
+                Some(rng.random())
             }
         })
         .collect()
@@ -86,10 +86,14 @@ pub fn create_month_day_nano_array_with_seed(
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
-                Some(IntervalMonthDayNano::new(rng.gen(), rng.gen(), rng.gen()))
+                Some(IntervalMonthDayNano::new(
+                    rng.random(),
+                    rng.random(),
+                    rng.random(),
+                ))
             }
         })
         .collect()
@@ -98,15 +102,15 @@ pub fn create_month_day_nano_array_with_seed(
 /// Creates a random (but fixed-seeded) array of a given size and null density
 pub fn create_boolean_array(size: usize, null_density: f32, true_density: f32) -> BooleanArray
 where
-    Standard: Distribution<bool>,
+    StandardUniform: Distribution<bool>,
 {
     let mut rng = seedable_rng();
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
-                let value = rng.gen::<f32>() < true_density;
+                let value = rng.random::<f32>() < true_density;
                 Some(value)
             }
         })
@@ -134,10 +138,10 @@ fn create_string_array_with_max_len<Offset: OffsetSizeTrait>(
     let rng = &mut seedable_rng();
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
-                let str_len = rng.gen_range(0..max_str_len);
+                let str_len = rng.random_range(0..max_str_len);
                 let value = rng.sample_iter(&Alphanumeric).take(str_len).collect();
                 let value = String::from_utf8(value).unwrap();
                 Some(value)
@@ -156,7 +160,7 @@ pub fn create_string_array_with_len<Offset: OffsetSizeTrait>(
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
                 let value = rng.sample_iter(&Alphanumeric).take(str_len).collect();
@@ -183,10 +187,10 @@ fn create_string_view_array_with_max_len(
     let rng = &mut seedable_rng();
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
-                let str_len = rng.gen_range(0..max_str_len);
+                let str_len = rng.random_range(0..max_str_len);
                 let value = rng.sample_iter(&Alphanumeric).take(str_len).collect();
                 let value = String::from_utf8(value).unwrap();
                 Some(value)
@@ -209,10 +213,10 @@ pub fn create_string_view_array_with_len(
     // if mixed, we creates first half that string length small than 12 bytes and second half large than 12 bytes
     if mixed {
         for _ in 0..size / 2 {
-            lengths.push(rng.gen_range(1..12));
+            lengths.push(rng.random_range(1..12));
         }
         for _ in size / 2..size {
-            lengths.push(rng.gen_range(12..=std::cmp::max(30, str_len)));
+            lengths.push(rng.random_range(12..=std::cmp::max(30, str_len)));
         }
     } else {
         lengths.resize(size, str_len);
@@ -221,7 +225,7 @@ pub fn create_string_view_array_with_len(
     lengths
         .into_iter()
         .map(|len| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
                 let value: Vec<u8> = rng.sample_iter(&Alphanumeric).take(len).collect();
@@ -242,7 +246,7 @@ pub fn create_string_dict_array<K: ArrowDictionaryKeyType>(
 
     let data: Vec<_> = (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
                 let value = rng.sample_iter(&Alphanumeric).take(str_len).collect();
@@ -296,7 +300,7 @@ pub fn create_string_array_for_runs(
     string_len: usize,
 ) -> Vec<String> {
     assert!(logical_array_len >= physical_array_len);
-    let mut rng = thread_rng();
+    let mut rng = rng();
 
     // typical length of each run
     let run_len = logical_array_len / physical_array_len;
@@ -305,7 +309,7 @@ pub fn create_string_array_for_runs(
     let mut run_len_extra = logical_array_len % physical_array_len;
 
     let mut values: Vec<String> = (0..physical_array_len)
-        .map(|_| (0..string_len).map(|_| rng.gen::<char>()).collect())
+        .map(|_| (0..string_len).map(|_| rng.random::<char>()).collect())
         .flat_map(|s| {
             let mut take_len = run_len;
             if run_len_extra > 0 {
@@ -332,12 +336,12 @@ pub fn create_binary_array<Offset: OffsetSizeTrait>(
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
                 let value = rng
-                    .sample_iter::<u8, _>(Standard)
-                    .take(range_rng.gen_range(0..8))
+                    .sample_iter::<u8, _>(StandardUniform)
+                    .take(range_rng.random_range(0..8))
                     .collect::<Vec<u8>>();
                 Some(value)
             }
@@ -351,11 +355,11 @@ pub fn create_fsb_array(size: usize, null_density: f32, value_len: usize) -> Fix
 
     FixedSizeBinaryArray::try_from_sparse_iter_with_size(
         (0..size).map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
                 let value = rng
-                    .sample_iter::<u8, _>(Standard)
+                    .sample_iter::<u8, _>(StandardUniform)
                     .take(value_len)
                     .collect::<Vec<u8>>();
                 Some(value)
@@ -375,7 +379,7 @@ pub fn create_dict_from_values<K>(
 ) -> DictionaryArray<K>
 where
     K: ArrowDictionaryKeyType,
-    Standard: Distribution<K::Native>,
+    StandardUniform: Distribution<K::Native>,
     K::Native: SampleUniform,
 {
     let min_key = K::Native::from_usize(0).unwrap();
@@ -393,7 +397,7 @@ pub fn create_sparse_dict_from_values<K>(
 ) -> DictionaryArray<K>
 where
     K: ArrowDictionaryKeyType,
-    Standard: Distribution<K::Native>,
+    StandardUniform: Distribution<K::Native>,
     K::Native: SampleUniform,
 {
     let mut rng = seedable_rng();
@@ -401,11 +405,14 @@ where
         DataType::Dictionary(Box::new(K::DATA_TYPE), Box::new(values.data_type().clone()));
 
     let keys: Buffer = (0..size)
-        .map(|_| rng.gen_range(key_range.clone()))
+        .map(|_| rng.random_range(key_range.clone()))
         .collect();
 
-    let nulls: Option<Buffer> =
-        (null_density != 0.).then(|| (0..size).map(|_| rng.gen_bool(null_density as _)).collect());
+    let nulls: Option<Buffer> = (null_density != 0.).then(|| {
+        (0..size)
+            .map(|_| rng.random_bool(null_density as _))
+            .collect()
+    });
 
     let data = ArrayDataBuilder::new(data_type)
         .len(size)
@@ -424,10 +431,10 @@ pub fn create_f16_array(size: usize, nan_density: f32) -> Float16Array {
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < nan_density {
+            if rng.random::<f32>() < nan_density {
                 Some(f16::NAN)
             } else {
-                Some(f16::from_f32(rng.gen()))
+                Some(f16::from_f32(rng.random()))
             }
         })
         .collect()
@@ -439,10 +446,10 @@ pub fn create_f32_array(size: usize, nan_density: f32) -> Float32Array {
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < nan_density {
+            if rng.random::<f32>() < nan_density {
                 Some(f32::NAN)
             } else {
-                Some(rng.gen())
+                Some(rng.random())
             }
         })
         .collect()
@@ -454,10 +461,10 @@ pub fn create_f64_array(size: usize, nan_density: f32) -> Float64Array {
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < nan_density {
+            if rng.random::<f32>() < nan_density {
                 Some(f64::NAN)
             } else {
-                Some(rng.gen())
+                Some(rng.random())
             }
         })
         .collect()
diff --git a/arrow/src/util/data_gen.rs b/arrow/src/util/data_gen.rs
index 5f63812e5..ee5350363 100644
--- a/arrow/src/util/data_gen.rs
+++ b/arrow/src/util/data_gen.rs
@@ -19,8 +19,10 @@
 
 use std::sync::Arc;
 
-use rand::distributions::uniform::SampleRange;
-use rand::{distributions::uniform::SampleUniform, Rng};
+use rand::{
+    distr::uniform::{SampleRange, SampleUniform},
+    Rng,
+};
 
 use crate::array::*;
 use crate::error::{ArrowError, Result};
@@ -370,7 +372,7 @@ fn create_random_offsets<T: OffsetSizeTrait + SampleUniform>(
     offsets.push(current_offset);
 
     (0..size).for_each(|_| {
-        current_offset += rng.gen_range(min..max);
+        current_offset += rng.random_range(min..max);
         offsets.push(current_offset);
     });
 
@@ -383,7 +385,7 @@ fn create_random_null_buffer(size: usize, null_density: f32) -> Buffer {
     {
         let mut_slice = mut_buf.as_slice_mut();
         (0..size).for_each(|i| {
-            if rng.gen::<f32>() >= null_density {
+            if rng.random::<f32>() >= null_density {
                 bit_util::set_bit(mut_slice, i)
             }
         })
@@ -402,7 +404,7 @@ pub trait RandomTemporalValue: ArrowTemporalType {
     where
         Self::Native: SampleUniform,
     {
-        rng.gen_range(Self::value_range())
+        rng.random_range(Self::value_range())
     }
 
     /// Generate a random value of the type
@@ -503,7 +505,7 @@ where
 
     (0..size)
         .map(|_| {
-            if rng.gen::<f32>() < null_density {
+            if rng.random::<f32>() < null_density {
                 None
             } else {
                 Some(T::random(&mut rng))
diff --git a/dev/release/release-tarball.sh b/dev/release/release-tarball.sh
index 352c15a30..0d111392f 100755
--- a/dev/release/release-tarball.sh
+++ b/dev/release/release-tarball.sh
@@ -75,5 +75,5 @@ echo "Success!"
 echo "The release is available here:"
 echo "  https://dist.apache.org/repos/dist/release/arrow/${release_version}"
 
-echo "Clean up old versions from svn"
-"${SOURCE_TOP_DIR}"/dev/release/remove-old-releases.sh
+echo "Clean up old artifacts from svn"
+"${SOURCE_TOP_DIR}"/dev/release/remove-old-artifacts.sh
diff --git a/dev/release/remove-old-releases.sh b/dev/release/remove-old-artifacts.sh
similarity index 61%
rename from dev/release/remove-old-releases.sh
rename to dev/release/remove-old-artifacts.sh
index 0722197a8..06bcc2740 100755
--- a/dev/release/remove-old-releases.sh
+++ b/dev/release/remove-old-artifacts.sh
@@ -18,8 +18,8 @@
 # under the License.
 #
 
-# This script removes all but the most recent versions of arrow-rs
-# from svn
+# This script removes all RCs and all but the most recent versions of
+# arrow-rs from svn.
 #
 # The older versions are in SVN history as well as available on the
 # archive page https://archive.apache.org/dist/
@@ -29,17 +29,35 @@
 
 set -e
 set -u
+set -o pipefail
 
-svn_base="https://dist.apache.org/repos/dist/release/arrow"
+echo "Remove all RCs"
+dev_base_url=https://dist.apache.org/repos/dist/dev/arrow
+old_rcs=$(
+  svn ls ${dev_base_url}/ | \
+  grep -E '^apache-arrow-rs-[0-9]' | \
+  sort --version-sort
+)
+for old_rc in $old_rcs; do
+  echo "Remove RC: ${old_rc}"
+  svn \
+    delete \
+    -m "Remove old Apache Arrow Rust RC: ${old_rc}" \
+    ${dev_base_url}/${old_rc}
+done
 
 echo "Remove all but the most recent version"
+release_base_url="https://dist.apache.org/repos/dist/release/arrow"
 old_releases=$(
-  svn ls ${svn_base} | \
+  svn ls ${release_base_url} | \
   grep -E '^arrow-rs-[0-9\.]+' | \
   sort --version-sort --reverse | \
   tail -n +2
 )
 for old_release_version in $old_releases; do
-  echo "Remove old release ${old_release_version}"
-  svn delete -m "Removing ${old_release_version}" ${svn_base}/${old_release_version}
+  echo "Remove old release: ${old_release_version}"
+  svn \
+    delete \
+    -m "Remove Apache Arrow Rust release: ${old_release_version}" \
+    ${release_base_url}/${old_release_version}
 done
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index 992ae6662..168d2eb6a 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -49,7 +49,7 @@ hyper = { version = "1.2", default-features = false, optional = true }
 quick-xml = { version = "0.37.0", features = ["serialize", "overlapped-lists"], optional = true }
 serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
 serde_json = { version = "1.0", default-features = false, optional = true }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"], optional = true }
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "http2"], optional = true }
 ring = { version = "0.17", default-features = false, features = ["std"], optional = true }
 rustls-pemfile = { version = "2.0", default-features = false, features = ["std"], optional = true }
@@ -76,7 +76,7 @@ futures-test = "0.3"
 hyper = { version = "1.2", features = ["server"] }
 hyper-util = "0.1"
 http-body-util = "0.1"
-rand = "0.8"
+rand = "0.9"
 tempfile = "3.1.0"
 regex = "1.11.1"
 # The "gzip" feature for reqwest is enabled for an integration test.
diff --git a/object_store/dev/release/release-tarball.sh b/object_store/dev/release/release-tarball.sh
index 958118639..16b10e022 100755
--- a/object_store/dev/release/release-tarball.sh
+++ b/object_store/dev/release/release-tarball.sh
@@ -75,5 +75,5 @@ echo "Success!"
 echo "The release is available here:"
 echo "  https://dist.apache.org/repos/dist/release/arrow/${release_version}"
 
-echo "Clean up old versions from svn"
-"${SOURCE_TOP_DIR}"/dev/release/remove-old-releases.sh
+echo "Clean up old artifacts from svn"
+"${SOURCE_TOP_DIR}"/dev/release/remove-old-artifacts.sh
diff --git a/object_store/dev/release/remove-old-releases.sh b/object_store/dev/release/remove-old-artifacts.sh
similarity index 60%
rename from object_store/dev/release/remove-old-releases.sh
rename to object_store/dev/release/remove-old-artifacts.sh
index c8bd8b748..bbbbe0c36 100755
--- a/object_store/dev/release/remove-old-releases.sh
+++ b/object_store/dev/release/remove-old-artifacts.sh
@@ -18,8 +18,8 @@
 # under the License.
 #
 
-# This script removes all but the most recent versions of arrow-rs
-# from svn
+# This script removes all RCs and all but the most recent versions of
+# object_store from svn.
 #
 # The older versions are in SVN history as well as available on the
 # archive page https://archive.apache.org/dist/
@@ -29,17 +29,35 @@
 
 set -e
 set -u
+set -o pipefail
 
-svn_base="https://dist.apache.org/repos/dist/release/arrow"
+echo "Remove all RCs"
+dev_base_url=https://dist.apache.org/repos/dist/dev/arrow
+old_rcs=$(
+  svn ls ${dev_base_url}/ | \
+  grep -E '^apache-arrow-object-store-rs-[0-9]' | \
+  sort --version-sort
+)
+for old_rc in $old_rcs; do
+  echo "Remove RC: ${old_rc}"
+  svn \
+    delete \
+    -m "Remove old Apache Arrow Rust Object Store RC: ${old_rc}" \
+    ${dev_base_url}/${old_rc}
+done
 
 echo "Remove all but the most recent version"
+release_base_url="https://dist.apache.org/repos/dist/release/arrow"
 old_releases=$(
-  svn ls ${svn_base} | \
+  svn ls ${release_base_url} | \
   grep -E '^arrow-object-store-rs-[0-9\.]+' | \
   sort --version-sort --reverse | \
   tail -n +2
 )
 for old_release_version in $old_releases; do
-  echo "Remove old release ${old_release_version}"
-  svn delete -m "Removing ${old_release_version}" ${svn_base}/${old_release_version}
+  echo "Remove old release: ${old_release_version}"
+  svn \
+    delete \
+    -m "Remove Apache Arrow Rust Object Store release: ${old_release_version}" \
+    ${release_base_url}/${old_release_version}
 done
diff --git a/object_store/src/aws/dynamo.rs b/object_store/src/aws/dynamo.rs
index 6283e76c1..a66a343be 100644
--- a/object_store/src/aws/dynamo.rs
+++ b/object_store/src/aws/dynamo.rs
@@ -527,8 +527,8 @@ mod tests {
     use super::*;
     use crate::aws::AmazonS3;
     use crate::ObjectStore;
-    use rand::distributions::Alphanumeric;
-    use rand::{thread_rng, Rng};
+    use rand::distr::Alphanumeric;
+    use rand::{rng, Rng};
 
     #[test]
     fn test_attribute_serde() {
@@ -571,7 +571,7 @@ mod tests {
             _ => panic!("Should conflict"),
         }
 
-        let rng = thread_rng();
+        let rng = rng();
         let etag = String::from_utf8(rng.sample_iter(Alphanumeric).take(32).collect()).unwrap();
         let t = Some(etag.as_str());
 
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 82ef909de..b065927b6 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -571,6 +571,7 @@ mod tests {
         stream_get(&integration).await;
         multipart(&integration, &integration).await;
         multipart_race_condition(&integration, true).await;
+        multipart_out_of_order(&integration).await;
         signing(&integration).await;
         s3_encryption(&integration).await;
         put_get_attributes(&integration).await;
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index 2c2e27ea4..7195729af 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -561,7 +561,7 @@ impl AzureClient {
         _part_idx: usize,
         payload: PutPayload,
     ) -> Result<PartId> {
-        let part_idx = u128::from_be_bytes(rand::thread_rng().gen());
+        let part_idx = u128::from_be_bytes(rand::rng().random());
         let content_id = format!("{part_idx:032x}");
         let block_id = BASE64_STANDARD.encode(&content_id);
 
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index ea4dd8f56..bbecba571 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -314,6 +314,7 @@ mod tests {
         put_opts(&integration, true).await;
         multipart(&integration, &integration).await;
         multipart_race_condition(&integration, false).await;
+        multipart_out_of_order(&integration).await;
         signing(&integration).await;
 
         let validate = !integration.client.config().disable_tagging;
diff --git a/object_store/src/client/backoff.rs b/object_store/src/client/backoff.rs
index 8382a2e16..8193e8bca 100644
--- a/object_store/src/client/backoff.rs
+++ b/object_store/src/client/backoff.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use rand::prelude::*;
+use rand::{prelude::*, rng};
 use std::time::Duration;
 
 /// Exponential backoff with decorrelated jitter algorithm
@@ -78,7 +78,7 @@ impl Backoff {
 
     /// Creates a new `Backoff` with the optional `rng`
     ///
-    /// Used [`rand::thread_rng()`] if no rng provided
+    /// Used [`rand::rng()`] if no rng provided
     pub(crate) fn new_with_rng(
         config: &BackoffConfig,
         rng: Option<Box<dyn RngCore + Sync + Send>>,
@@ -98,8 +98,8 @@ impl Backoff {
         let range = self.init_backoff..(self.next_backoff_secs * self.base);
 
         let rand_backoff = match self.rng.as_mut() {
-            Some(rng) => rng.gen_range(range),
-            None => thread_rng().gen_range(range),
+            Some(rng) => rng.random_range(range),
+            None => rng().random_range(range),
         };
 
         let next_backoff = self.max_backoff_secs.min(rand_backoff);
diff --git a/object_store/src/gcp/mod.rs b/object_store/src/gcp/mod.rs
index a2f512415..2aa99765d 100644
--- a/object_store/src/gcp/mod.rs
+++ b/object_store/src/gcp/mod.rs
@@ -298,6 +298,7 @@ mod test {
             stream_get(&integration).await;
             multipart(&integration, &integration).await;
             multipart_race_condition(&integration, true).await;
+            multipart_out_of_order(&integration).await;
             // Fake GCS server doesn't currently honor preconditions
             get_opts(&integration).await;
             put_opts(&integration, true).await;
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
index 25a929459..f10b2d31e 100644
--- a/object_store/src/integration.rs
+++ b/object_store/src/integration.rs
@@ -29,14 +29,14 @@ use core::str;
 use crate::multipart::MultipartStore;
 use crate::path::Path;
 use crate::{
-    Attribute, Attributes, DynObjectStore, Error, GetOptions, GetRange, ObjectStore, PutMode,
-    PutPayload, UpdateVersion, WriteMultipart,
+    Attribute, Attributes, DynObjectStore, Error, GetOptions, GetRange, MultipartUpload,
+    ObjectStore, PutMode, PutPayload, UpdateVersion, WriteMultipart,
 };
 use bytes::Bytes;
 use futures::stream::FuturesUnordered;
 use futures::{StreamExt, TryStreamExt};
-use rand::distributions::Alphanumeric;
-use rand::{thread_rng, Rng};
+use rand::distr::Alphanumeric;
+use rand::{rng, Rng};
 
 pub(crate) async fn flatten_list_stream(
     storage: &DynObjectStore,
@@ -633,7 +633,7 @@ pub async fn put_opts(storage: &dyn ObjectStore, supports_update: bool) {
     // As a result each conditional operation will need to wait for the lease to timeout before proceeding
     // One solution would be to clear DynamoDB before each test, but this would require non-trivial additional code
     // so we instead just generate a random suffix for the filenames
-    let rng = thread_rng();
+    let rng = rng();
     let suffix = String::from_utf8(rng.sample_iter(Alphanumeric).take(32).collect()).unwrap();
 
     delete_fixtures(storage).await;
@@ -742,10 +742,10 @@ pub async fn put_opts(storage: &dyn ObjectStore, supports_update: bool) {
 /// Returns a chunk of length `chunk_length`
 fn get_chunk(chunk_length: usize) -> Bytes {
     let mut data = vec![0_u8; chunk_length];
-    let mut rng = thread_rng();
+    let mut rng = rng();
     // Set a random selection of bytes
     for _ in 0..1000 {
-        data[rng.gen_range(0..chunk_length)] = rng.gen();
+        data[rng.random_range(0..chunk_length)] = rng.random();
     }
     data.into()
 }
@@ -1196,3 +1196,34 @@ pub async fn multipart_race_condition(storage: &dyn ObjectStore, last_writer_win
         ));
     }
 }
+
+/// Tests performing out of order multipart uploads
+pub async fn multipart_out_of_order(storage: &dyn ObjectStore) {
+    let path = Path::from("test_multipart_out_of_order");
+    let mut multipart_upload = storage.put_multipart(&path).await.unwrap();
+
+    let part1 = std::iter::repeat(b'1')
+        .take(5 * 1024 * 1024)
+        .collect::<Bytes>();
+    let part2 = std::iter::repeat(b'2')
+        .take(5 * 1024 * 1024)
+        .collect::<Bytes>();
+    let part3 = std::iter::repeat(b'3')
+        .take(5 * 1024 * 1024)
+        .collect::<Bytes>();
+    let full = [part1.as_ref(), part2.as_ref(), part3.as_ref()].concat();
+
+    let fut1 = multipart_upload.put_part(part1.into());
+    let fut2 = multipart_upload.put_part(part2.into());
+    let fut3 = multipart_upload.put_part(part3.into());
+    // note order is 2,3,1 , different than the parts were created in
+    fut2.await.unwrap();
+    fut3.await.unwrap();
+    fut1.await.unwrap();
+
+    multipart_upload.complete().await.unwrap();
+
+    let result = storage.get(&path).await.unwrap();
+    let bytes = result.bytes().await.unwrap();
+    assert_eq!(bytes, full);
+}
diff --git a/object_store/src/upload.rs b/object_store/src/upload.rs
index 4df4d8fd4..af5975a74 100644
--- a/object_store/src/upload.rs
+++ b/object_store/src/upload.rs
@@ -312,11 +312,11 @@ mod tests {
                     let mut expected = Vec::with_capacity(1024);
 
                     for _ in 0..50 {
-                        let chunk_size = rng.gen_range(0..30);
-                        let data: Vec<_> = (0..chunk_size).map(|_| rng.gen()).collect();
+                        let chunk_size = rng.random_range(0..30);
+                        let data: Vec<_> = (0..chunk_size).map(|_| rng.random()).collect();
                         expected.extend_from_slice(&data);
 
-                        match rng.gen_bool(method) {
+                        match rng.random_bool(method) {
                             true => write.put(data.into()),
                             false => write.write(&data),
                         }
diff --git a/object_store/src/util.rs b/object_store/src/util.rs
index 17a7a8cad..f46c959ba 100644
--- a/object_store/src/util.rs
+++ b/object_store/src/util.rs
@@ -329,7 +329,7 @@ mod tests {
     use crate::Error;
 
     use super::*;
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
     use std::ops::Range;
 
     /// Calls coalesce_ranges and validates the returned data is correct
@@ -395,20 +395,20 @@ mod tests {
 
     #[tokio::test]
     async fn test_coalesce_fuzz() {
-        let mut rand = thread_rng();
+        let mut rand = rng();
         for _ in 0..100 {
-            let object_len = rand.gen_range(10..250);
-            let range_count = rand.gen_range(0..10);
+            let object_len = rand.random_range(10..250);
+            let range_count = rand.random_range(0..10);
             let ranges: Vec<_> = (0..range_count)
                 .map(|_| {
-                    let start = rand.gen_range(0..object_len);
+                    let start = rand.random_range(0..object_len);
                     let max_len = 20.min(object_len - start);
-                    let len = rand.gen_range(0..max_len);
+                    let len = rand.random_range(0..max_len);
                     start..start + len
                 })
                 .collect();
 
-            let coalesce = rand.gen_range(1..5);
+            let coalesce = rand.random_range(1..5);
             let fetches = do_fetch(ranges.clone(), coalesce).await;
 
             for fetch in fetches.windows(2) {
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 54992d864..75cd867ed 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -83,7 +83,7 @@ zstd = { version = "0.13", default-features = false }
 serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] }
-rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
+rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 object_store = { version = "0.11.0", default-features = false, features = ["azure"] }
 
 # TODO: temporary to fix parquet wasm build
@@ -103,6 +103,8 @@ default = ["arrow", "snap", "brotli", "flate2", "lz4", "zstd", "base64", "simdut
 lz4 = ["lz4_flex"]
 # Enable arrow reader/writer APIs
 arrow = ["base64", "arrow-array", "arrow-buffer", "arrow-cast", "arrow-data", "arrow-schema", "arrow-select", "arrow-ipc"]
+# Enable support for arrow canonical extension types
+arrow_canonical_extension_types = ["arrow-schema?/canonical_extension_types"]
 # Enable CLI tools
 cli = ["json", "base64", "clap", "arrow-csv", "serde"]
 # Enable JSON APIs
diff --git a/parquet/benches/arrow_reader.rs b/parquet/benches/arrow_reader.rs
index e5165fee2..0e887c31c 100644
--- a/parquet/benches/arrow_reader.rs
+++ b/parquet/benches/arrow_reader.rs
@@ -37,7 +37,7 @@ use parquet::{
     data_type::{ByteArrayType, Int32Type, Int64Type},
     schema::types::{ColumnDescPtr, SchemaDescPtr},
 };
-use rand::distributions::uniform::SampleUniform;
+use rand::distr::uniform::SampleUniform;
 use rand::{rngs::StdRng, Rng, SeedableRng};
 use std::{collections::VecDeque, sync::Arc};
 
@@ -119,14 +119,14 @@ where
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for _k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
                 };
                 if def_level == max_def_level {
                     // create the Float16 value
-                    let value = f16::from_f32(rng.gen_range(min..max));
+                    let value = f16::from_f32(rng.random_range(min..max));
                     // Float16 in parquet is stored little-endian
                     let bytes = match column_desc.physical_type() {
                         Type::FIXED_LEN_BYTE_ARRAY => {
@@ -177,14 +177,14 @@ where
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for _k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
                 };
                 if def_level == max_def_level {
                     // create the decimal value
-                    let value = rng.gen_range(min..max);
+                    let value = rng.random_range(min..max);
                     // decimal of parquet use the big-endian to store
                     let bytes = match column_desc.physical_type() {
                         Type::BYTE_ARRAY => {
@@ -235,14 +235,14 @@ fn build_encoded_flba_bytes_page_iterator<const BYTE_LENGTH: usize>(
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for _k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
                 };
                 if def_level == max_def_level {
                     // create the FLBA(BYTE_LENGTH) value
-                    let value = (0..BYTE_LENGTH).map(|_| rng.gen()).collect::<Vec<u8>>();
+                    let value = (0..BYTE_LENGTH).map(|_| rng.random()).collect::<Vec<u8>>();
                     let value =
                         <FixedLenByteArrayType as parquet::data_type::DataType>::T::from(value);
                     values.push(value);
@@ -284,13 +284,13 @@ where
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for _k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
                 };
                 if def_level == max_def_level {
-                    let value = FromPrimitive::from_usize(rng.gen_range(min..max)).unwrap();
+                    let value = FromPrimitive::from_usize(rng.random_range(min..max)).unwrap();
                     values.push(value);
                 }
                 def_levels.push(def_level);
@@ -336,14 +336,14 @@ where
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for _k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
                 };
                 if def_level == max_def_level {
                     // select random value from list of unique values
-                    let value = unique_values[rng.gen_range(0..NUM_UNIQUE_VALUES)];
+                    let value = unique_values[rng.random_range(0..NUM_UNIQUE_VALUES)];
                     values.push(value);
                 }
                 def_levels.push(def_level);
@@ -393,7 +393,7 @@ fn build_plain_encoded_byte_array_page_iterator_inner(
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
@@ -452,14 +452,15 @@ fn build_dictionary_encoded_string_page_iterator(
             let mut values = Vec::with_capacity(VALUES_PER_PAGE);
             let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
             for _k in 0..VALUES_PER_PAGE {
-                let def_level = if rng.gen::<f32>() < null_density {
+                let def_level = if rng.random::<f32>() < null_density {
                     max_def_level - 1
                 } else {
                     max_def_level
                 };
                 if def_level == max_def_level {
                     // select random value from list of unique values
-                    let string_value = unique_values[rng.gen_range(0..NUM_UNIQUE_VALUES)].as_str();
+                    let string_value =
+                        unique_values[rng.random_range(0..NUM_UNIQUE_VALUES)].as_str();
                     values.push(parquet::data_type::ByteArray::from(string_value));
                 }
                 def_levels.push(def_level);
@@ -512,12 +513,12 @@ fn build_string_list_page_iterator(
             let mut rep_levels = Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
             for k in 0..VALUES_PER_PAGE {
                 rep_levels.push(0);
-                if rng.gen::<f32>() < null_density {
+                if rng.random::<f32>() < null_density {
                     // Null list
                     def_levels.push(0);
                     continue;
                 }
-                let len = rng.gen_range(0..MAX_LIST_LEN);
+                let len = rng.random_range(0..MAX_LIST_LEN);
                 if len == 0 {
                     // Empty list
                     def_levels.push(1);
@@ -527,7 +528,7 @@ fn build_string_list_page_iterator(
                 (1..len).for_each(|_| rep_levels.push(1));
 
                 for l in 0..len {
-                    if rng.gen::<f32>() < null_density {
+                    if rng.random::<f32>() < null_density {
                         // Null element
                         def_levels.push(2);
                     } else {
diff --git a/parquet/benches/compression.rs b/parquet/benches/compression.rs
index 2275a8940..5c9b0d1f7 100644
--- a/parquet/benches/compression.rs
+++ b/parquet/benches/compression.rs
@@ -18,7 +18,7 @@
 use criterion::*;
 use parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel};
 use parquet::compression::create_codec;
-use rand::distributions::Alphanumeric;
+use rand::distr::Alphanumeric;
 use rand::prelude::*;
 
 fn do_bench(c: &mut Criterion, name: &str, uncompressed: &[u8]) {
@@ -76,7 +76,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     // Create a collection of 64 words
     let words: Vec<Vec<_>> = (0..64)
         .map(|_| {
-            let len = rng.gen_range(1..12);
+            let len = rng.random_range(1..12);
             rng.sample_iter(&Alphanumeric).take(len).collect()
         })
         .collect();
@@ -84,7 +84,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     // Build data by concatenating these words randomly together
     let mut uncompressed = Vec::with_capacity(DATA_SIZE);
     while uncompressed.len() < DATA_SIZE {
-        let word = &words[rng.gen_range(0..words.len())];
+        let word = &words[rng.random_range(0..words.len())];
         uncompressed.extend_from_slice(&word[..word.len().min(DATA_SIZE - uncompressed.len())])
     }
     assert_eq!(uncompressed.len(), DATA_SIZE);
diff --git a/parquet/benches/encoding.rs b/parquet/benches/encoding.rs
index 8e61666e6..68f215d4e 100644
--- a/parquet/benches/encoding.rs
+++ b/parquet/benches/encoding.rs
@@ -86,12 +86,12 @@ fn criterion_benchmark(c: &mut Criterion) {
     let mut d128s = Vec::new();
     for _ in 0..n {
         f16s.push(FixedLenByteArray::from(
-            f16::from_f32(rng.gen::<f32>()).to_le_bytes().to_vec(),
+            f16::from_f32(rng.random::<f32>()).to_le_bytes().to_vec(),
         ));
-        f32s.push(rng.gen::<f32>());
-        f64s.push(rng.gen::<f64>());
+        f32s.push(rng.random::<f32>());
+        f64s.push(rng.random::<f64>());
         d128s.push(FixedLenByteArray::from(
-            rng.gen::<i128>().to_be_bytes().to_vec(),
+            rng.random::<i128>().to_be_bytes().to_vec(),
         ));
     }
 
diff --git a/parquet/benches/row_selector.rs b/parquet/benches/row_selector.rs
index 32f0d6a56..9fa634d09 100644
--- a/parquet/benches/row_selector.rs
+++ b/parquet/benches/row_selector.rs
@@ -31,9 +31,9 @@ use rand::Rng;
 ///
 /// * A `BooleanArray` instance with randomly selected rows based on the provided ratio.
 fn generate_random_row_selection(total_rows: usize, selection_ratio: f64) -> BooleanArray {
-    let mut rng = rand::thread_rng();
+    let mut rng = rand::rng();
     let bools: Vec<bool> = (0..total_rows)
-        .map(|_| rng.gen_bool(selection_ratio))
+        .map(|_| rng.random_bool(selection_ratio))
         .collect();
     BooleanArray::from(bools)
 }
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 8df659060..6d6bbdc7b 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -329,7 +329,7 @@ impl ByteViewArrayDecoderPlain {
 
         let to_read = len.min(self.max_remaining_values);
 
-        let buf = self.buf.as_ref();
+        let buf: &[u8] = self.buf.as_ref();
         let mut read = 0;
         output.views.reserve(to_read);
 
@@ -405,7 +405,7 @@ impl ByteViewArrayDecoderPlain {
     pub fn skip(&mut self, to_skip: usize) -> Result<usize> {
         let to_skip = to_skip.min(self.max_remaining_values);
         let mut skip = 0;
-        let buf = self.buf.as_ref();
+        let buf: &[u8] = self.buf.as_ref();
 
         while self.offset < self.buf.len() && skip != to_skip {
             if self.offset + 4 > buf.len() {
diff --git a/parquet/src/arrow/array_reader/primitive_array.rs b/parquet/src/arrow/array_reader/primitive_array.rs
index 709d0f8bb..c76c41db3 100644
--- a/parquet/src/arrow/array_reader/primitive_array.rs
+++ b/parquet/src/arrow/array_reader/primitive_array.rs
@@ -361,7 +361,7 @@ mod tests {
     use arrow_array::{Array, Date32Array, PrimitiveArray};
 
     use arrow::datatypes::DataType::{Date32, Decimal128};
-    use rand::distributions::uniform::SampleUniform;
+    use rand::distr::uniform::SampleUniform;
     use std::collections::VecDeque;
 
     #[allow(clippy::too_many_arguments)]
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index 6eba04c86..acdf078b2 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -317,7 +317,7 @@ impl ArrowReaderOptions {
     ///
     /// // Create the reader and read the data using the supplied schema.
     /// let mut reader = builder.build().unwrap();
-    /// let _batch = reader.next().unwrap().unwrap();   
+    /// let _batch = reader.next().unwrap().unwrap();
     /// ```
     pub fn with_schema(self, schema: SchemaRef) -> Self {
         Self {
@@ -926,7 +926,7 @@ mod tests {
     use bytes::Bytes;
     use half::f16;
     use num::PrimInt;
-    use rand::{thread_rng, Rng, RngCore};
+    use rand::{rng, Rng, RngCore};
     use tempfile::tempfile;
 
     use arrow_array::builder::*;
@@ -1403,7 +1403,7 @@ mod tests {
     impl RandGen<FixedLenByteArrayType> for RandFixedLenGen {
         fn gen(len: i32) -> FixedLenByteArray {
             let mut v = vec![0u8; len as usize];
-            thread_rng().fill_bytes(&mut v);
+            rng().fill_bytes(&mut v);
             ByteArray::from(v).into()
         }
     }
@@ -2058,10 +2058,13 @@ mod tests {
         fn with_row_selections(self) -> Self {
             assert!(self.row_filter.is_none(), "Must set row selection first");
 
-            let mut rng = thread_rng();
-            let step = rng.gen_range(self.record_batch_size..self.num_rows);
-            let row_selections =
-                create_test_selection(step, self.num_row_groups * self.num_rows, rng.gen::<bool>());
+            let mut rng = rng();
+            let step = rng.random_range(self.record_batch_size..self.num_rows);
+            let row_selections = create_test_selection(
+                step,
+                self.num_row_groups * self.num_rows,
+                rng.random::<bool>(),
+            );
             Self {
                 row_selections: Some(row_selections),
                 ..self
@@ -2074,9 +2077,9 @@ mod tests {
                 None => self.num_row_groups * self.num_rows,
             };
 
-            let mut rng = thread_rng();
+            let mut rng = rng();
             Self {
-                row_filter: Some((0..row_count).map(|_| rng.gen_bool(0.9)).collect()),
+                row_filter: Some((0..row_count).map(|_| rng.random_bool(0.9)).collect()),
                 ..self
             }
         }
@@ -2290,7 +2293,7 @@ mod tests {
         //according to null_percent generate def_levels
         let (repetition, def_levels) = match opts.null_percent.as_ref() {
             Some(null_percent) => {
-                let mut rng = thread_rng();
+                let mut rng = rng();
 
                 let def_levels: Vec<Vec<i16>> = (0..opts.num_row_groups)
                     .map(|_| {
@@ -4134,7 +4137,7 @@ mod tests {
 
     #[test]
     fn test_list_selection_fuzz() {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         let schema = Arc::new(Schema::new(vec![Field::new_list(
             "list",
             Field::new_list(
@@ -4150,26 +4153,26 @@ mod tests {
         let mut list_a_builder = ListBuilder::new(ListBuilder::new(Int32Builder::new()));
 
         for _ in 0..2048 {
-            if rng.gen_bool(0.2) {
+            if rng.random_bool(0.2) {
                 list_a_builder.append(false);
                 continue;
             }
 
-            let list_a_len = rng.gen_range(0..10);
+            let list_a_len = rng.random_range(0..10);
             let list_b_builder = list_a_builder.values();
 
             for _ in 0..list_a_len {
-                if rng.gen_bool(0.2) {
+                if rng.random_bool(0.2) {
                     list_b_builder.append(false);
                     continue;
                 }
 
-                let list_b_len = rng.gen_range(0..10);
+                let list_b_len = rng.random_range(0..10);
                 let int_builder = list_b_builder.values();
                 for _ in 0..list_b_len {
-                    match rng.gen_bool(0.2) {
+                    match rng.random_bool(0.2) {
                         true => int_builder.append_null(),
-                        false => int_builder.append_value(rng.gen()),
+                        false => int_builder.append_value(rng.random()),
                     }
                 }
                 list_b_builder.append(true)
diff --git a/parquet/src/arrow/arrow_reader/selection.rs b/parquet/src/arrow/arrow_reader/selection.rs
index 378d2253f..ffcf39df0 100644
--- a/parquet/src/arrow/arrow_reader/selection.rs
+++ b/parquet/src/arrow/arrow_reader/selection.rs
@@ -641,7 +641,7 @@ fn union_row_selections(left: &[RowSelector], right: &[RowSelector]) -> RowSelec
 mod tests {
     use super::*;
     use crate::format::PageLocation;
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
 
     #[test]
     fn test_from_filters() {
@@ -1013,14 +1013,14 @@ mod tests {
 
     #[test]
     fn test_and_fuzz() {
-        let mut rand = thread_rng();
+        let mut rand = rng();
         for _ in 0..100 {
-            let a_len = rand.gen_range(10..100);
-            let a_bools: Vec<_> = (0..a_len).map(|_| rand.gen_bool(0.2)).collect();
+            let a_len = rand.random_range(10..100);
+            let a_bools: Vec<_> = (0..a_len).map(|_| rand.random_bool(0.2)).collect();
             let a = RowSelection::from_filters(&[BooleanArray::from(a_bools.clone())]);
 
             let b_len: usize = a_bools.iter().map(|x| *x as usize).sum();
-            let b_bools: Vec<_> = (0..b_len).map(|_| rand.gen_bool(0.8)).collect();
+            let b_bools: Vec<_> = (0..b_len).map(|_| rand.random_bool(0.8)).collect();
             let b = RowSelection::from_filters(&[BooleanArray::from(b_bools.clone())]);
 
             let mut expected_bools = vec![false; a_len];
diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
index 2c8a59399..25fdb0266 100644
--- a/parquet/src/arrow/async_reader/mod.rs
+++ b/parquet/src/arrow/async_reader/mod.rs
@@ -1070,7 +1070,7 @@ mod tests {
     };
     use arrow_schema::{DataType, Field, Schema};
     use futures::{StreamExt, TryStreamExt};
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
     use std::collections::HashMap;
     use std::sync::{Arc, Mutex};
     use tempfile::tempfile;
@@ -1400,7 +1400,7 @@ mod tests {
 
         assert_eq!(metadata.num_row_groups(), 1);
 
-        let mut rand = thread_rng();
+        let mut rand = rng();
 
         for _ in 0..100 {
             let mut expected_rows = 0;
@@ -1409,7 +1409,7 @@ mod tests {
             let mut selectors = vec![];
 
             while total_rows < 7300 {
-                let row_count: usize = rand.gen_range(1..100);
+                let row_count: usize = rand.random_range(1..100);
 
                 let row_count = row_count.min(7300 - total_rows);
 
@@ -1436,7 +1436,7 @@ mod tests {
                 .await
                 .unwrap();
 
-            let col_idx: usize = rand.gen_range(0..13);
+            let col_idx: usize = rand.random_range(0..13);
             let mask = ProjectionMask::leaves(builder.parquet_schema(), vec![col_idx]);
 
             let stream = builder
@@ -1467,7 +1467,7 @@ mod tests {
 
         assert_eq!(metadata.num_row_groups(), 1);
 
-        let mut rand = thread_rng();
+        let mut rand = rng();
 
         let mut expected_rows = 0;
         let mut total_rows = 0;
@@ -1480,7 +1480,7 @@ mod tests {
         });
 
         while total_rows < 7300 {
-            let row_count: usize = rand.gen_range(1..100);
+            let row_count: usize = rand.random_range(1..100);
 
             let row_count = row_count.min(7300 - total_rows);
 
@@ -1507,7 +1507,7 @@ mod tests {
             .await
             .unwrap();
 
-        let col_idx: usize = rand.gen_range(0..13);
+        let col_idx: usize = rand.random_range(0..13);
         let mask = ProjectionMask::leaves(builder.parquet_schema(), vec![col_idx]);
 
         let stream = builder
diff --git a/parquet/src/arrow/buffer/bit_util.rs b/parquet/src/arrow/buffer/bit_util.rs
index e7aea56a7..1d2c953ab 100644
--- a/parquet/src/arrow/buffer/bit_util.rs
+++ b/parquet/src/arrow/buffer/bit_util.rs
@@ -65,12 +65,12 @@ pub fn sign_extend_be<const N: usize>(b: &[u8]) -> [u8; N] {
 mod tests {
     use super::*;
     use arrow_array::builder::BooleanBufferBuilder;
-    use rand::prelude::*;
+    use rand::{prelude::*, rng};
 
     #[test]
     fn test_bit_fns() {
-        let mut rng = thread_rng();
-        let mask_length = rng.gen_range(1..1024);
+        let mut rng = rng();
+        let mask_length = rng.random_range(1..1024);
         let bools: Vec<_> = std::iter::from_fn(|| Some(rng.next_u32() & 1 == 0))
             .take(mask_length)
             .collect();
@@ -92,8 +92,8 @@ mod tests {
         assert_eq!(count_set_bits(&[0xFF], 1..1), 0);
 
         for _ in 0..20 {
-            let start = rng.gen_range(0..bools.len());
-            let end = rng.gen_range(start..bools.len());
+            let start = rng.random_range(0..bools.len());
+            let end = rng.random_range(start..bools.len());
 
             let actual = count_set_bits(nulls.as_slice(), start..end);
             let expected = bools[start..end].iter().filter(|x| **x).count();
diff --git a/parquet/src/arrow/record_reader/definition_levels.rs b/parquet/src/arrow/record_reader/definition_levels.rs
index fcd04fbb9..a90b3c4ec 100644
--- a/parquet/src/arrow/record_reader/definition_levels.rs
+++ b/parquet/src/arrow/record_reader/definition_levels.rs
@@ -351,17 +351,17 @@ mod tests {
     use super::*;
 
     use crate::encodings::rle::RleEncoder;
-    use rand::{thread_rng, Rng};
+    use rand::{rng, Rng};
 
     #[test]
     fn test_packed_decoder() {
-        let mut rng = thread_rng();
-        let len: usize = rng.gen_range(512..1024);
+        let mut rng = rng();
+        let len: usize = rng.random_range(512..1024);
 
         let mut expected = BooleanBufferBuilder::new(len);
         let mut encoder = RleEncoder::new(1, 1024);
         for _ in 0..len {
-            let bool = rng.gen_bool(0.8);
+            let bool = rng.random_bool(0.8);
             encoder.put(bool as u64);
             expected.append(bool);
         }
@@ -379,7 +379,7 @@ mod tests {
                 break;
             }
 
-            let to_read = rng.gen_range(1..=remaining);
+            let to_read = rng.random_range(1..=remaining);
             decoder.read(&mut decoded, to_read).unwrap();
         }
 
@@ -389,15 +389,15 @@ mod tests {
 
     #[test]
     fn test_packed_decoder_skip() {
-        let mut rng = thread_rng();
-        let len: usize = rng.gen_range(512..1024);
+        let mut rng = rng();
+        let len: usize = rng.random_range(512..1024);
 
         let mut expected = BooleanBufferBuilder::new(len);
         let mut encoder = RleEncoder::new(1, 1024);
 
         let mut total_value = 0;
         for _ in 0..len {
-            let bool = rng.gen_bool(0.8);
+            let bool = rng.random_bool(0.8);
             encoder.put(bool as u64);
             expected.append(bool);
             if bool {
@@ -421,8 +421,8 @@ mod tests {
             if remaining_levels == 0 {
                 break;
             }
-            let to_read_or_skip_level = rng.gen_range(1..=remaining_levels);
-            if rng.gen_bool(0.5) {
+            let to_read_or_skip_level = rng.random_range(1..=remaining_levels);
+            if rng.random_bool(0.5) {
                 let (skip_val_num, skip_level_num) = decoder.skip(to_read_or_skip_level).unwrap();
                 skip_value += skip_val_num;
                 skip_level += skip_level_num
diff --git a/parquet/src/arrow/schema/mod.rs b/parquet/src/arrow/schema/mod.rs
index 8be243900..89c42f5ea 100644
--- a/parquet/src/arrow/schema/mod.rs
+++ b/parquet/src/arrow/schema/mod.rs
@@ -23,6 +23,8 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use arrow_ipc::writer;
+#[cfg(feature = "arrow_canonical_extension_types")]
+use arrow_schema::extension::{Json, Uuid};
 use arrow_schema::{DataType, Field, Fields, Schema, TimeUnit};
 
 use crate::basic::{
@@ -380,12 +382,26 @@ pub fn parquet_to_arrow_field(parquet_column: &ColumnDescriptor) -> Result<Field
     let mut ret = Field::new(parquet_column.name(), field.arrow_type, field.nullable);
 
     let basic_info = parquet_column.self_type().get_basic_info();
+    let mut meta = HashMap::with_capacity(if cfg!(feature = "arrow_canonical_extension_types") {
+        2
+    } else {
+        1
+    });
     if basic_info.has_id() {
-        let mut meta = HashMap::with_capacity(1);
         meta.insert(
             PARQUET_FIELD_ID_META_KEY.to_string(),
             basic_info.id().to_string(),
         );
+    }
+    #[cfg(feature = "arrow_canonical_extension_types")]
+    if let Some(logical_type) = basic_info.logical_type() {
+        match logical_type {
+            LogicalType::Uuid => ret.try_with_extension_type(Uuid)?,
+            LogicalType::Json => ret.try_with_extension_type(Json::default())?,
+            _ => {}
+        }
+    }
+    if !meta.is_empty() {
         ret.set_metadata(meta);
     }
 
@@ -590,6 +606,16 @@ fn arrow_to_parquet_type(field: &Field, coerce_types: bool) -> Result<Type> {
                 .with_repetition(repetition)
                 .with_id(id)
                 .with_length(*length)
+                .with_logical_type(
+                    #[cfg(feature = "arrow_canonical_extension_types")]
+                    // If set, map arrow uuid extension type to parquet uuid logical type.
+                    field
+                        .try_extension_type::<Uuid>()
+                        .ok()
+                        .map(|_| LogicalType::Uuid),
+                    #[cfg(not(feature = "arrow_canonical_extension_types"))]
+                    None,
+                )
                 .build()
         }
         DataType::BinaryView => Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
@@ -623,13 +649,35 @@ fn arrow_to_parquet_type(field: &Field, coerce_types: bool) -> Result<Type> {
         }
         DataType::Utf8 | DataType::LargeUtf8 => {
             Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
-                .with_logical_type(Some(LogicalType::String))
+                .with_logical_type({
+                    #[cfg(feature = "arrow_canonical_extension_types")]
+                    {
+                        // Use the Json logical type if the canonical Json
+                        // extension type is set on this field.
+                        field
+                            .try_extension_type::<Json>()
+                            .map_or(Some(LogicalType::String), |_| Some(LogicalType::Json))
+                    }
+                    #[cfg(not(feature = "arrow_canonical_extension_types"))]
+                    Some(LogicalType::String)
+                })
                 .with_repetition(repetition)
                 .with_id(id)
                 .build()
         }
         DataType::Utf8View => Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
-            .with_logical_type(Some(LogicalType::String))
+            .with_logical_type({
+                #[cfg(feature = "arrow_canonical_extension_types")]
+                {
+                    // Use the Json logical type if the canonical Json
+                    // extension type is set on this field.
+                    field
+                        .try_extension_type::<Json>()
+                        .map_or(Some(LogicalType::String), |_| Some(LogicalType::Json))
+                }
+                #[cfg(not(feature = "arrow_canonical_extension_types"))]
+                Some(LogicalType::String)
+            })
             .with_repetition(repetition)
             .with_id(id)
             .build(),
@@ -2163,4 +2211,53 @@ mod tests {
     fn test_get_arrow_schema_from_metadata() {
         assert!(get_arrow_schema_from_metadata("").is_err());
     }
+
+    #[test]
+    #[cfg(feature = "arrow_canonical_extension_types")]
+    fn arrow_uuid_to_parquet_uuid() -> Result<()> {
+        let arrow_schema = Schema::new(vec![Field::new(
+            "uuid",
+            DataType::FixedSizeBinary(16),
+            false,
+        )
+        .with_extension_type(Uuid)]);
+
+        let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;
+
+        assert_eq!(
+            parquet_schema.column(0).logical_type(),
+            Some(LogicalType::Uuid)
+        );
+
+        // TODO: roundtrip
+        // let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
+        // assert_eq!(arrow_schema.field(0).try_extension_type::<Uuid>()?, Uuid);
+
+        Ok(())
+    }
+
+    #[test]
+    #[cfg(feature = "arrow_canonical_extension_types")]
+    fn arrow_json_to_parquet_json() -> Result<()> {
+        let arrow_schema = Schema::new(vec![
+            Field::new("json", DataType::Utf8, false).with_extension_type(Json::default())
+        ]);
+
+        let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;
+
+        assert_eq!(
+            parquet_schema.column(0).logical_type(),
+            Some(LogicalType::Json)
+        );
+
+        // TODO: roundtrip
+        // https://github.com/apache/arrow-rs/issues/7063
+        // let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
+        // assert_eq!(
+        //     arrow_schema.field(0).try_extension_type::<Json>()?,
+        //     Json::default()
+        // );
+
+        Ok(())
+    }
 }
diff --git a/parquet/src/column/reader.rs b/parquet/src/column/reader.rs
index 953dc057d..b69980578 100644
--- a/parquet/src/column/reader.rs
+++ b/parquet/src/column/reader.rs
@@ -252,7 +252,7 @@ where
                     let (values_read, levels_read) = reader.read_def_levels(out, levels_to_read)?;
 
                     if levels_read != levels_to_read {
-                        return Err(general_err!("insufficient definition levels read from column - expected {rep_levels}, got {read}"));
+                        return Err(general_err!("insufficient definition levels read from column - expected {levels_to_read}, got {levels_read}"));
                     }
 
                     values_read
@@ -583,7 +583,7 @@ fn parse_v1_level(
 mod tests {
     use super::*;
 
-    use rand::distributions::uniform::SampleUniform;
+    use rand::distr::uniform::SampleUniform;
     use std::{collections::VecDeque, sync::Arc};
 
     use crate::basic::Type as PhysicalType;
diff --git a/parquet/src/column/reader/decoder.rs b/parquet/src/column/reader/decoder.rs
index afd58b3cd..a8766e821 100644
--- a/parquet/src/column/reader/decoder.rs
+++ b/parquet/src/column/reader/decoder.rs
@@ -484,7 +484,7 @@ impl RepetitionLevelDecoder for RepetitionLevelDecoderImpl {
 mod tests {
     use super::*;
     use crate::encodings::rle::RleEncoder;
-    use rand::prelude::*;
+    use rand::{prelude::*, rng};
 
     #[test]
     fn test_skip_padding() {
@@ -509,9 +509,9 @@ mod tests {
     #[test]
     fn test_skip_rep_levels() {
         for _ in 0..10 {
-            let mut rng = thread_rng();
+            let mut rng = rng();
             let total_len = 10000_usize;
-            let mut encoded: Vec<i16> = (0..total_len).map(|_| rng.gen_range(0..5)).collect();
+            let mut encoded: Vec<i16> = (0..total_len).map(|_| rng.random_range(0..5)).collect();
             encoded[0] = 0;
             let mut encoder = RleEncoder::new(3, 1024);
             for v in &encoded {
@@ -526,8 +526,8 @@ mod tests {
             let mut remaining_records = total_records;
             let mut remaining_levels = encoded.len();
             loop {
-                let skip = rng.gen_bool(0.5);
-                let records = rng.gen_range(1..=remaining_records.min(5));
+                let skip = rng.random_bool(0.5);
+                let records = rng.random_range(1..=remaining_records.min(5));
                 let (records_read, levels_read) = if skip {
                     decoder.skip_rep_levels(records, remaining_levels).unwrap()
                 } else {
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 5f34f34cb..ae4182375 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -1528,7 +1528,7 @@ mod tests {
         schema::parser::parse_message_type,
     };
     use core::str;
-    use rand::distributions::uniform::SampleUniform;
+    use rand::distr::uniform::SampleUniform;
     use std::{fs::File, sync::Arc};
 
     use crate::column::{
diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index d089ba783..3fb0ebb84 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -528,7 +528,7 @@ mod tests {
     use super::*;
 
     use crate::util::bit_util::ceil;
-    use rand::{self, distributions::Standard, thread_rng, Rng, SeedableRng};
+    use rand::{self, distr::StandardUniform, rng, Rng, SeedableRng};
 
     const MAX_WIDTH: usize = 32;
 
@@ -1019,15 +1019,18 @@ mod tests {
 
         for _ in 0..niters {
             values.clear();
-            let rng = thread_rng();
-            let seed_vec: Vec<u8> = rng.sample_iter::<u8, _>(&Standard).take(seed_len).collect();
+            let rng = rng();
+            let seed_vec: Vec<u8> = rng
+                .sample_iter::<u8, _>(&StandardUniform)
+                .take(seed_len)
+                .collect();
             let mut seed = [0u8; 32];
             seed.copy_from_slice(&seed_vec[0..seed_len]);
             let mut gen = rand::rngs::StdRng::from_seed(seed);
 
             let mut parity = false;
             for _ in 0..ngroups {
-                let mut group_size = gen.gen_range(1..20);
+                let mut group_size = gen.random_range(1..20);
                 if group_size > max_group_size {
                     group_size = 1;
                 }
diff --git a/parquet/src/util/bit_util.rs b/parquet/src/util/bit_util.rs
index b4c929ce0..8f6c2d8f8 100644
--- a/parquet/src/util/bit_util.rs
+++ b/parquet/src/util/bit_util.rs
@@ -716,7 +716,7 @@ mod tests {
     use super::*;
 
     use crate::util::test_common::rand_gen::random_numbers;
-    use rand::distributions::{Distribution, Standard};
+    use rand::distr::{Distribution, StandardUniform};
     use std::fmt::Debug;
 
     #[test]
@@ -1066,7 +1066,7 @@ mod tests {
     fn test_put_aligned_rand_numbers<T>(total: usize, num_bits: usize)
     where
         T: Copy + FromBytes + AsBytes + Debug + PartialEq,
-        Standard: Distribution<T>,
+        StandardUniform: Distribution<T>,
     {
         assert!(num_bits <= 32);
         assert!(total % 2 == 0);


Test patch:
diff --git a/arrow/src/util/test_util.rs b/arrow/src/util/test_util.rs
index 2d718d392..566ccc6ab 100644
--- a/arrow/src/util/test_util.rs
+++ b/arrow/src/util/test_util.rs
@@ -25,7 +25,7 @@ pub fn random_bytes(n: usize) -> Vec<u8> {
     let mut result = vec![];
     let mut rng = seedable_rng();
     for _ in 0..n {
-        result.push(rng.gen_range(0..255));
+        result.push(rng.random_range(0..255));
     }
     result
 }
diff --git a/parquet/src/util/test_common/rand_gen.rs b/parquet/src/util/test_common/rand_gen.rs
index ec80d3a59..36a7e4b52 100644
--- a/parquet/src/util/test_common/rand_gen.rs
+++ b/parquet/src/util/test_common/rand_gen.rs
@@ -19,8 +19,8 @@ use crate::basic::Encoding;
 use crate::column::page::Page;
 use bytes::Bytes;
 use rand::{
-    distributions::{uniform::SampleUniform, Distribution, Standard},
-    thread_rng, Rng,
+    distr::{uniform::SampleUniform, Distribution, StandardUniform},
+    rng, Rng,
 };
 use std::collections::VecDeque;
 
@@ -44,51 +44,55 @@ pub trait RandGen<T: DataType> {
 
 impl RandGen<BoolType> for BoolType {
     fn gen(_: i32) -> bool {
-        thread_rng().gen::<bool>()
+        rng().random::<bool>()
     }
 }
 
 impl RandGen<Int32Type> for Int32Type {
     fn gen(_: i32) -> i32 {
-        thread_rng().gen::<i32>()
+        rng().random::<i32>()
     }
 }
 
 impl RandGen<Int64Type> for Int64Type {
     fn gen(_: i32) -> i64 {
-        thread_rng().gen::<i64>()
+        rng().random::<i64>()
     }
 }
 
 impl RandGen<Int96Type> for Int96Type {
     fn gen(_: i32) -> Int96 {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         let mut result = Int96::new();
-        result.set_data(rng.gen::<u32>(), rng.gen::<u32>(), rng.gen::<u32>());
+        result.set_data(
+            rng.random::<u32>(),
+            rng.random::<u32>(),
+            rng.random::<u32>(),
+        );
         result
     }
 }
 
 impl RandGen<FloatType> for FloatType {
     fn gen(_: i32) -> f32 {
-        thread_rng().gen::<f32>()
+        rng().random::<f32>()
     }
 }
 
 impl RandGen<DoubleType> for DoubleType {
     fn gen(_: i32) -> f64 {
-        thread_rng().gen::<f64>()
+        rng().random::<f64>()
     }
 }
 
 impl RandGen<ByteArrayType> for ByteArrayType {
     fn gen(_: i32) -> ByteArray {
-        let mut rng = thread_rng();
+        let mut rng = rng();
         let mut result = ByteArray::new();
         let mut value = vec![];
-        let len = rng.gen_range(0..128);
+        let len = rng.random_range(0..128);
         for _ in 0..len {
-            value.push(rng.gen_range(0..255));
+            value.push(rng.random_range(0..255));
         }
         result.set_data(Bytes::from(value));
         result
@@ -105,28 +109,28 @@ impl RandGen<FixedLenByteArrayType> for FixedLenByteArrayType {
 
 pub fn random_bytes(n: usize) -> Vec<u8> {
     let mut result = vec![];
-    let mut rng = thread_rng();
+    let mut rng = rng();
     for _ in 0..n {
-        result.push(rng.gen_range(0..255));
+        result.push(rng.random_range(0..255));
     }
     result
 }
 
 pub fn random_numbers<T>(n: usize) -> Vec<T>
 where
-    Standard: Distribution<T>,
+    StandardUniform: Distribution<T>,
 {
-    let mut rng = thread_rng();
-    Standard.sample_iter(&mut rng).take(n).collect()
+    let mut rng = rng();
+    StandardUniform.sample_iter(&mut rng).take(n).collect()
 }
 
 pub fn random_numbers_range<T>(n: usize, low: T, high: T, result: &mut Vec<T>)
 where
     T: PartialOrd + SampleUniform + Copy,
 {
-    let mut rng = thread_rng();
+    let mut rng = rng();
     for _ in 0..n {
-        result.push(rng.gen_range(low..high));
+        result.push(rng.random_range(low..high));
     }
 }