Repository: apache/arrow-rs
Pull request #6259 (2024-08-15 19:34:28)

Add `Statistics::distinct_count_opt` and deprecate `Statistics::distinct_count`
# Which issue does this PR close?


# Rationale for this change

in https://github.com/apache/arrow-rs/pull/6216 from @Michael-J-Ward most of the methods on `Statistics` where deprecated and replaced with `_opt` variants

 For example `Statistics::null_count` is now deprecated and replaced with `Statistics::null_count_opt`

 However, for whatever reason `Statistics::distinct_count` already returned an Option and thus `Statistics::distinct_count_opt` was not added in https://github.com/apache/arrow-rs/pull/6216

This is now a papercut in my opinion as the API is not consistent. Access to `distinct_count` should be via `distinct_count_opt` as well




# What changes are included in this PR?
* Add `Statistics::distinct_count_opt` and deprecate `Statistics::distinct_count`


# Are there any user-facing changes?
* deprecated method

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/README.md b/README.md
index 944e8562b..a52549084 100644
--- a/README.md
+++ b/README.md
@@ -92,12 +92,15 @@ There are several related crates in different repositories
 | [`datafusion`]           | In-memory query engine with SQL support     | [(README)][datafusion-readme]           |
 | [`ballista`]             | Distributed query execution                 | [(README)][ballista-readme]             |
 | [`object_store_opendal`] | Use [`opendal`] as [`object_store`] backend | [(README)][object_store_opendal-readme] |
+| [`parquet_opendal`]      | Use [`opendal`] for [`parquet`] Arrow IO    | [(README)][parquet_opendal-readme]      |
 
 [`datafusion`]: https://crates.io/crates/datafusion
 [`ballista`]: https://crates.io/crates/ballista
 [`object_store_opendal`]: https://crates.io/crates/object_store_opendal
 [`opendal`]: https://crates.io/crates/opendal
 [object_store_opendal-readme]: https://github.com/apache/opendal/blob/main/integrations/object_store/README.md
+[`parquet_opendal`]: https://crates.io/crates/parquet_opendal
+[parquet_opendal-readme]: https://github.com/apache/opendal/blob/main/integrations/parquet/README.md
 
 Collectively, these crates support a wider array of functionality for analytic computations in Rust.
 
diff --git a/arrow-arith/src/temporal.rs b/arrow-arith/src/temporal.rs
index 6f8e8cc18..5f3eeb325 100644
--- a/arrow-arith/src/temporal.rs
+++ b/arrow-arith/src/temporal.rs
@@ -32,7 +32,7 @@ use arrow_array::timezone::Tz;
 use arrow_array::types::*;
 use arrow_array::*;
 use arrow_buffer::ArrowNativeType;
-use arrow_schema::{ArrowError, DataType, IntervalUnit};
+use arrow_schema::{ArrowError, DataType, IntervalUnit, TimeUnit};
 
 /// Valid parts to extract from date/time/timestamp arrays.
 ///
@@ -113,6 +113,7 @@ where
 ///   - Time32/Time64
 ///   - Timestamp
 ///   - Interval
+///   - Duration
 ///
 /// Returns an [`Int32Array`] unless input was a dictionary type, in which case returns
 /// the dictionary but with this function applied onto its values.
@@ -154,6 +155,26 @@ pub fn date_part(array: &dyn Array, part: DatePart) -> Result<ArrayRef, ArrowErr
             let array = Arc::new(array) as ArrayRef;
             Ok(array)
         }
+        DataType::Duration(TimeUnit::Second) => {
+            let array = as_primitive_array::<DurationSecondType>(array).date_part(part)?;
+            let array = Arc::new(array) as ArrayRef;
+            Ok(array)
+        }
+        DataType::Duration(TimeUnit::Millisecond) => {
+            let array = as_primitive_array::<DurationMillisecondType>(array).date_part(part)?;
+            let array = Arc::new(array) as ArrayRef;
+            Ok(array)
+        }
+        DataType::Duration(TimeUnit::Microsecond) => {
+            let array = as_primitive_array::<DurationMicrosecondType>(array).date_part(part)?;
+            let array = Arc::new(array) as ArrayRef;
+            Ok(array)
+        }
+        DataType::Duration(TimeUnit::Nanosecond) => {
+            let array = as_primitive_array::<DurationNanosecondType>(array).date_part(part)?;
+            let array = Arc::new(array) as ArrayRef;
+            Ok(array)
+        }
         DataType::Dictionary(_, _) => {
             let array = array.as_any_dictionary();
             let values = date_part(array.values(), part)?;
@@ -482,6 +503,126 @@ impl ExtractDatePartExt for PrimitiveArray<IntervalMonthDayNanoType> {
     }
 }
 
+impl ExtractDatePartExt for PrimitiveArray<DurationSecondType> {
+    fn date_part(&self, part: DatePart) -> Result<Int32Array, ArrowError> {
+        match part {
+            DatePart::Week => Ok(self.unary_opt(|d| (d / (60 * 60 * 24 * 7)).try_into().ok())),
+            DatePart::Day => Ok(self.unary_opt(|d| (d / (60 * 60 * 24)).try_into().ok())),
+            DatePart::Hour => Ok(self.unary_opt(|d| (d / (60 * 60)).try_into().ok())),
+            DatePart::Minute => Ok(self.unary_opt(|d| (d / 60).try_into().ok())),
+            DatePart::Second => Ok(self.unary_opt(|d| d.try_into().ok())),
+            DatePart::Millisecond => {
+                Ok(self.unary_opt(|d| d.checked_mul(1_000).and_then(|d| d.try_into().ok())))
+            }
+            DatePart::Microsecond => {
+                Ok(self.unary_opt(|d| d.checked_mul(1_000_000).and_then(|d| d.try_into().ok())))
+            }
+            DatePart::Nanosecond => Ok(
+                self.unary_opt(|d| d.checked_mul(1_000_000_000).and_then(|d| d.try_into().ok()))
+            ),
+
+            DatePart::Year
+            | DatePart::Quarter
+            | DatePart::Month
+            | DatePart::DayOfWeekSunday0
+            | DatePart::DayOfWeekMonday0
+            | DatePart::DayOfYear => {
+                return_compute_error_with!(format!("{part} does not support"), self.data_type())
+            }
+        }
+    }
+}
+
+impl ExtractDatePartExt for PrimitiveArray<DurationMillisecondType> {
+    fn date_part(&self, part: DatePart) -> Result<Int32Array, ArrowError> {
+        match part {
+            DatePart::Week => {
+                Ok(self.unary_opt(|d| (d / (1_000 * 60 * 60 * 24 * 7)).try_into().ok()))
+            }
+            DatePart::Day => Ok(self.unary_opt(|d| (d / (1_000 * 60 * 60 * 24)).try_into().ok())),
+            DatePart::Hour => Ok(self.unary_opt(|d| (d / (1_000 * 60 * 60)).try_into().ok())),
+            DatePart::Minute => Ok(self.unary_opt(|d| (d / (1_000 * 60)).try_into().ok())),
+            DatePart::Second => Ok(self.unary_opt(|d| (d / 1_000).try_into().ok())),
+            DatePart::Millisecond => Ok(self.unary_opt(|d| d.try_into().ok())),
+            DatePart::Microsecond => {
+                Ok(self.unary_opt(|d| d.checked_mul(1_000).and_then(|d| d.try_into().ok())))
+            }
+            DatePart::Nanosecond => {
+                Ok(self.unary_opt(|d| d.checked_mul(1_000_000).and_then(|d| d.try_into().ok())))
+            }
+
+            DatePart::Year
+            | DatePart::Quarter
+            | DatePart::Month
+            | DatePart::DayOfWeekSunday0
+            | DatePart::DayOfWeekMonday0
+            | DatePart::DayOfYear => {
+                return_compute_error_with!(format!("{part} does not support"), self.data_type())
+            }
+        }
+    }
+}
+
+impl ExtractDatePartExt for PrimitiveArray<DurationMicrosecondType> {
+    fn date_part(&self, part: DatePart) -> Result<Int32Array, ArrowError> {
+        match part {
+            DatePart::Week => {
+                Ok(self.unary_opt(|d| (d / (1_000_000 * 60 * 60 * 24 * 7)).try_into().ok()))
+            }
+            DatePart::Day => {
+                Ok(self.unary_opt(|d| (d / (1_000_000 * 60 * 60 * 24)).try_into().ok()))
+            }
+            DatePart::Hour => Ok(self.unary_opt(|d| (d / (1_000_000 * 60 * 60)).try_into().ok())),
+            DatePart::Minute => Ok(self.unary_opt(|d| (d / (1_000_000 * 60)).try_into().ok())),
+            DatePart::Second => Ok(self.unary_opt(|d| (d / 1_000_000).try_into().ok())),
+            DatePart::Millisecond => Ok(self.unary_opt(|d| (d / 1_000).try_into().ok())),
+            DatePart::Microsecond => Ok(self.unary_opt(|d| d.try_into().ok())),
+            DatePart::Nanosecond => {
+                Ok(self.unary_opt(|d| d.checked_mul(1_000).and_then(|d| d.try_into().ok())))
+            }
+
+            DatePart::Year
+            | DatePart::Quarter
+            | DatePart::Month
+            | DatePart::DayOfWeekSunday0
+            | DatePart::DayOfWeekMonday0
+            | DatePart::DayOfYear => {
+                return_compute_error_with!(format!("{part} does not support"), self.data_type())
+            }
+        }
+    }
+}
+
+impl ExtractDatePartExt for PrimitiveArray<DurationNanosecondType> {
+    fn date_part(&self, part: DatePart) -> Result<Int32Array, ArrowError> {
+        match part {
+            DatePart::Week => {
+                Ok(self.unary_opt(|d| (d / (1_000_000_000 * 60 * 60 * 24 * 7)).try_into().ok()))
+            }
+            DatePart::Day => {
+                Ok(self.unary_opt(|d| (d / (1_000_000_000 * 60 * 60 * 24)).try_into().ok()))
+            }
+            DatePart::Hour => {
+                Ok(self.unary_opt(|d| (d / (1_000_000_000 * 60 * 60)).try_into().ok()))
+            }
+            DatePart::Minute => Ok(self.unary_opt(|d| (d / (1_000_000_000 * 60)).try_into().ok())),
+            DatePart::Second => Ok(self.unary_opt(|d| (d / 1_000_000_000).try_into().ok())),
+            DatePart::Millisecond => Ok(self.unary_opt(|d| (d / 1_000_000).try_into().ok())),
+            DatePart::Microsecond => Ok(self.unary_opt(|d| (d / 1_000).try_into().ok())),
+            DatePart::Nanosecond => Ok(self.unary_opt(|d| d.try_into().ok())),
+
+            DatePart::Year
+            | DatePart::Quarter
+            | DatePart::Month
+            | DatePart::DayOfWeekSunday0
+            | DatePart::DayOfWeekMonday0
+            | DatePart::DayOfYear => {
+                return_compute_error_with!(format!("{part} does not support"), self.data_type())
+            }
+        }
+    }
+}
+
 macro_rules! return_compute_error_with {
     ($msg:expr, $param:expr) => {
         return { Err(ArrowError::ComputeError(format!("{}: {:?}", $msg, $param))) }
@@ -1796,4 +1937,148 @@ mod tests {
             IntervalMonthDayNano::ZERO,
         ]));
     }
+
+    #[test]
+    fn test_duration_second() {
+        let input: DurationSecondArray = vec![0, 42, 60 * 60 * 24 + 1].into();
+
+        let actual = date_part(&input, DatePart::Second).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42, actual.value(1));
+        assert_eq!(60 * 60 * 24 + 1, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Millisecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42_000, actual.value(1));
+        assert_eq!((60 * 60 * 24 + 1) * 1_000, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Microsecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42_000_000, actual.value(1));
+        assert_eq!(0, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Nanosecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!(0, actual.value(2));
+    }
+
+    #[test]
+    fn test_duration_millisecond() {
+        let input: DurationMillisecondArray = vec![0, 42, 60 * 60 * 24 + 1].into();
+
+        let actual = date_part(&input, DatePart::Second).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!((60 * 60 * 24 + 1) / 1_000, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Millisecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42, actual.value(1));
+        assert_eq!(60 * 60 * 24 + 1, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Microsecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42_000, actual.value(1));
+        assert_eq!((60 * 60 * 24 + 1) * 1_000, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Nanosecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42_000_000, actual.value(1));
+        assert_eq!(0, actual.value(2));
+    }
+
+    #[test]
+    fn test_duration_microsecond() {
+        let input: DurationMicrosecondArray = vec![0, 42, 60 * 60 * 24 + 1].into();
+
+        let actual = date_part(&input, DatePart::Second).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!(0, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Millisecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!((60 * 60 * 24 + 1) / 1_000, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Microsecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42, actual.value(1));
+        assert_eq!(60 * 60 * 24 + 1, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Nanosecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42_000, actual.value(1));
+        assert_eq!((60 * 60 * 24 + 1) * 1_000, actual.value(2));
+    }
+
+    #[test]
+    fn test_duration_nanosecond() {
+        let input: DurationNanosecondArray = vec![0, 42, 60 * 60 * 24 + 1].into();
+
+        let actual = date_part(&input, DatePart::Second).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!(0, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Millisecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!(0, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Microsecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(0, actual.value(1));
+        assert_eq!((60 * 60 * 24 + 1) / 1_000, actual.value(2));
+
+        let actual = date_part(&input, DatePart::Nanosecond).unwrap();
+        let actual = actual.as_primitive::<Int32Type>();
+        assert_eq!(0, actual.value(0));
+        assert_eq!(42, actual.value(1));
+        assert_eq!(60 * 60 * 24 + 1, actual.value(2));
+    }
+
+    #[test]
+    fn test_duration_invalid_parts() {
+        fn ensure_returns_error(array: &dyn Array) {
+            let invalid_parts = [
+                DatePart::Year,
+                DatePart::Quarter,
+                DatePart::Month,
+                DatePart::DayOfWeekSunday0,
+                DatePart::DayOfWeekMonday0,
+                DatePart::DayOfYear,
+            ];
+
+            for part in invalid_parts {
+                let err = date_part(array, part).unwrap_err();
+                let expected = format!(
+                    "Compute error: {part} does not support: {}",
+                    array.data_type()
+                );
+                assert_eq!(expected, err.to_string());
+            }
+        }
+
+        ensure_returns_error(&DurationSecondArray::from(vec![0]));
+        ensure_returns_error(&DurationMillisecondArray::from(vec![0]));
+        ensure_returns_error(&DurationMicrosecondArray::from(vec![0]));
+        ensure_returns_error(&DurationNanosecondArray::from(vec![0]));
+    }
 }
diff --git a/arrow-data/src/transform/mod.rs b/arrow-data/src/transform/mod.rs
index 1e43bf488..c6d834aea 100644
--- a/arrow-data/src/transform/mod.rs
+++ b/arrow-data/src/transform/mod.rs
@@ -97,40 +97,78 @@ fn build_extend_null_bits(array: &ArrayData, use_nulls: bool) -> ExtendNullBits
     }
 }
 
-/// Struct to efficiently and interactively create an [ArrayData] from an existing [ArrayData] by
+/// Efficiently create an [ArrayData] from one or more existing [ArrayData]s by
 /// copying chunks.
 ///
-/// The main use case of this struct is to perform unary operations to arrays of arbitrary types,
-/// such as `filter` and `take`.
+/// The main use case of this struct is to perform unary operations to arrays of
+/// arbitrary types, such as `filter` and `take`.
+///
+/// # Example
+/// ```
+/// use arrow_buffer::Buffer;
+/// use arrow_data::ArrayData;
+/// use arrow_data::transform::MutableArrayData;
+/// use arrow_schema::DataType;
+/// fn i32_array(values: &[i32]) -> ArrayData {
+///   ArrayData::try_new(DataType::Int32, 5, None, 0, vec![Buffer::from_slice_ref(values)], vec![]).unwrap()
+/// }
+/// let arr1  = i32_array(&[1, 2, 3, 4, 5]);
+/// let arr2  = i32_array(&[6, 7, 8, 9, 10]);
+/// // Create a mutable array for copying values from arr1 and arr2, with a capacity for 6 elements
+/// let capacity = 3 * size_of::<i32>();
+/// let mut mutable = MutableArrayData::new(vec![&arr1, &arr2], false, 10);
+/// // Copy the first 3 elements from arr1
+/// mutable.extend(0, 0, 3);
+/// // Copy the last 3 elements from arr2
+/// mutable.extend(1, 2, 4);
+/// // Complete the MutableArrayData into a new ArrayData
+/// let frozen = mutable.freeze();
+/// assert_eq!(frozen, i32_array(&[1, 2, 3, 8, 9, 10]));
+/// ```
 pub struct MutableArrayData<'a> {
+    /// Input arrays: the data being read FROM.
+    ///
+    /// Note this is "dead code" because all actual references to the arrays are
+    /// stored in closures for extending values and nulls.
     #[allow(dead_code)]
     arrays: Vec<&'a ArrayData>,
-    /// The attributes in [_MutableArrayData] cannot be in [MutableArrayData] due to
-    /// mutability invariants (interior mutability):
-    /// [MutableArrayData] contains a function that can only mutate [_MutableArrayData], not
-    /// [MutableArrayData] itself
+
+    /// In progress output array: The data being written TO
+    ///
+    /// Note these fields are in a separate struct, [_MutableArrayData], as they
+    /// cannot be in [MutableArrayData] itself due to mutability invariants (interior
+    /// mutability): [MutableArrayData] contains a function that can only mutate
+    /// [_MutableArrayData], not [MutableArrayData] itself
     data: _MutableArrayData<'a>,
 
-    /// the child data of the `Array` in Dictionary arrays.
-    /// This is not stored in `MutableArrayData` because these values constant and only needed
-    /// at the end, when freezing [_MutableArrayData].
+    /// The child data of the `Array` in Dictionary arrays.
+    ///
+    /// This is not stored in `_MutableArrayData` because these values are
+    /// constant and only needed at the end, when freezing [_MutableArrayData].
     dictionary: Option<ArrayData>,
 
-    /// Variadic data buffers referenced by views
-    /// This is not stored in `MutableArrayData` because these values constant  and only needed
-    /// at the end, when freezing [_MutableArrayData]
+    /// Variadic data buffers referenced by views.
+    ///
+    /// Note this this is not stored in `_MutableArrayData` because these values
+    /// are constant and only needed at the end, when freezing
+    /// [_MutableArrayData]
     variadic_data_buffers: Vec<Buffer>,
 
-    /// function used to extend values from arrays. This function's lifetime is bound to the array
-    /// because it reads values from it.
+    /// function used to extend output array with values from input arrays.
+    ///
+    /// This function's lifetime is bound to the input arrays because it reads
+    /// values from them.
     extend_values: Vec<Extend<'a>>,
 
-    /// function used to extend nulls from arrays. This function's lifetime is bound to the array
-    /// because it reads nulls from it.
+    /// function used to extend the output array with nulls from input arrays.
+    ///
+    /// This function's lifetime is bound to the input arrays because it reads
+    /// nulls from it.
     extend_null_bits: Vec<ExtendNullBits<'a>>,
 
-    /// function used to extend nulls.
-    /// this is independent of the arrays and therefore has no lifetime.
+    /// function used to extend the output array with null elements.
+    ///
+    /// This function is independent of the arrays and therefore has no lifetime.
     extend_nulls: ExtendNulls,
 }
 
@@ -307,47 +345,63 @@ fn preallocate_offset_and_binary_buffer<Offset: ArrowNativeType + Integer>(
     ]
 }
 
-/// Define capacities of child data or data buffers.
+/// Define capacities to pre-allocate for child data or data buffers.
 #[derive(Debug, Clone)]
 pub enum Capacities {
     /// Binary, Utf8 and LargeUtf8 data types
-    /// Define
+    ///
+    /// Defines
     /// * the capacity of the array offsets
     /// * the capacity of the binary/ str buffer
     Binary(usize, Option<usize>),
     /// List and LargeList data types
-    /// Define
+    ///
+    /// Defines
     /// * the capacity of the array offsets
     /// * the capacity of the child data
     List(usize, Option<Box<Capacities>>),
     /// Struct type
+    ///
+    /// Defines
     /// * the capacity of the array
     /// * the capacities of the fields
     Struct(usize, Option<Vec<Capacities>>),
     /// Dictionary type
+    ///
+    /// Defines
     /// * the capacity of the array/keys
     /// * the capacity of the values
     Dictionary(usize, Option<Box<Capacities>>),
     /// Don't preallocate inner buffers and rely on array growth strategy
     Array(usize),
 }
+
 impl<'a> MutableArrayData<'a> {
-    /// returns a new [MutableArrayData] with capacity to `capacity` slots and specialized to create an
-    /// [ArrayData] from multiple `arrays`.
+    /// Returns a new [MutableArrayData] with capacity to `capacity` slots and
+    /// specialized to create an [ArrayData] from multiple `arrays`.
     ///
-    /// `use_nulls` is a flag used to optimize insertions. It should be `false` if the only source of nulls
-    /// are the arrays themselves and `true` if the user plans to call [MutableArrayData::extend_nulls].
-    /// In other words, if `use_nulls` is `false`, calling [MutableArrayData::extend_nulls] should not be used.
+    /// # Arguments
+    /// * `arrays` - the source arrays to copy from
+    /// * `use_nulls` - a flag used to optimize insertions
+    ///   - `false` if the only source of nulls are the arrays themselves
+    ///   - `true` if the user plans to call [MutableArrayData::extend_nulls].
+    /// * capacity - the preallocated capacity of the output array, in bytes
+    ///
+    /// Thus, if `use_nulls` is `false`, calling
+    /// [MutableArrayData::extend_nulls] should not be used.
     pub fn new(arrays: Vec<&'a ArrayData>, use_nulls: bool, capacity: usize) -> Self {
         Self::with_capacities(arrays, use_nulls, Capacities::Array(capacity))
     }
 
-    /// Similar to [MutableArrayData::new], but lets users define the preallocated capacities of the array.
-    /// See also [MutableArrayData::new] for more information on the arguments.
+    /// Similar to [MutableArrayData::new], but lets users define the
+    /// preallocated capacities of the array with more granularity.
     ///
-    /// # Panic
-    /// This function panics if the given `capacities` don't match the data type of `arrays`. Or when
-    /// a [Capacities] variant is not yet supported.
+    /// See [MutableArrayData::new] for more information on the arguments.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if the given `capacities` don't match the data type
+    /// of `arrays`. Or when a [Capacities] variant is not yet supported.
     pub fn with_capacities(
         arrays: Vec<&'a ArrayData>,
         use_nulls: bool,
@@ -646,7 +700,7 @@ impl<'a> MutableArrayData<'a> {
         }
     }
 
-    /// Extends this array with a chunk of its source arrays
+    /// Extends the in progress array with a region of the input arrays
     ///
     /// # Arguments
     /// * `index` - the index of array that you what to copy values from
@@ -664,12 +718,11 @@ impl<'a> MutableArrayData<'a> {
         self.data.len += len;
     }
 
-    /// Extends this [MutableArrayData] with null elements, disregarding the bound arrays
+    /// Extends the in progress array with null elements, ignoring the input arrays.
     ///
     /// # Panics
     ///
     /// Panics if [`MutableArrayData`] not created with `use_nulls` or nullable source arrays
-    ///
     pub fn extend_nulls(&mut self, len: usize) {
         self.data.len += len;
         let bit_len = bit_util::ceil(self.data.len, 8);
@@ -697,12 +750,13 @@ impl<'a> MutableArrayData<'a> {
         self.data.null_count
     }
 
-    /// Creates a [ArrayData] from the pushed regions up to this point, consuming `self`.
+    /// Creates a [ArrayData] from the in progress array, consuming `self`.
     pub fn freeze(self) -> ArrayData {
         unsafe { self.into_builder().build_unchecked() }
     }
 
-    /// Creates a [ArrayDataBuilder] from the pushed regions up to this point, consuming `self`.
+    /// Consume self and returns the in progress array as [`ArrayDataBuilder`].
+    ///
     /// This is useful for extending the default behavior of MutableArrayData.
     pub fn into_builder(self) -> ArrayDataBuilder {
         let data = self.data;
diff --git a/arrow-flight/examples/flight_sql_server.rs b/arrow-flight/examples/flight_sql_server.rs
index d5168debc..81afecf85 100644
--- a/arrow-flight/examples/flight_sql_server.rs
+++ b/arrow-flight/examples/flight_sql_server.rs
@@ -46,9 +46,9 @@ use arrow_flight::sql::{
     ActionEndTransactionRequest, Any, CommandGetCatalogs, CommandGetCrossReference,
     CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys, CommandGetPrimaryKeys,
     CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables, CommandGetXdbcTypeInfo,
-    CommandPreparedStatementQuery, CommandPreparedStatementUpdate, CommandStatementQuery,
-    CommandStatementSubstraitPlan, CommandStatementUpdate, Nullable, ProstMessageExt, Searchable,
-    SqlInfo, TicketStatementQuery, XdbcDataType,
+    CommandPreparedStatementQuery, CommandPreparedStatementUpdate, CommandStatementIngest,
+    CommandStatementQuery, CommandStatementSubstraitPlan, CommandStatementUpdate, Nullable,
+    ProstMessageExt, Searchable, SqlInfo, TicketStatementQuery, XdbcDataType,
 };
 use arrow_flight::utils::batches_to_flight_data;
 use arrow_flight::{
@@ -615,6 +615,14 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(FAKE_UPDATE_RESULT)
     }
 
+    async fn do_put_statement_ingest(
+        &self,
+        _ticket: CommandStatementIngest,
+        _request: Request<PeekableFlightDataStream>,
+    ) -> Result<i64, Status> {
+        Ok(FAKE_UPDATE_RESULT)
+    }
+
     async fn do_put_substrait_plan(
         &self,
         _ticket: CommandStatementSubstraitPlan,
diff --git a/arrow-flight/src/client.rs b/arrow-flight/src/client.rs
index 3f62b256d..af3c8fba3 100644
--- a/arrow-flight/src/client.rs
+++ b/arrow-flight/src/client.rs
@@ -679,7 +679,7 @@ impl FlightClient {
 /// it encounters an error it uses the oneshot sender to
 /// notify the error and stop any further streaming. See `do_put` or
 /// `do_exchange` for it's uses.
-struct FallibleRequestStream<T, E> {
+pub(crate) struct FallibleRequestStream<T, E> {
     /// sender to notify error
     sender: Option<Sender<E>>,
     /// fallible stream
@@ -687,7 +687,7 @@ struct FallibleRequestStream<T, E> {
 }
 
 impl<T, E> FallibleRequestStream<T, E> {
-    fn new(
+    pub(crate) fn new(
         sender: Sender<E>,
         fallible_stream: Pin<Box<dyn Stream<Item = std::result::Result<T, E>> + Send + 'static>>,
     ) -> Self {
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index 91790898b..345254a63 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -24,6 +24,7 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use tonic::metadata::AsciiMetadataKey;
 
+use crate::client::FallibleRequestStream;
 use crate::decode::FlightRecordBatchStream;
 use crate::encode::FlightDataEncoderBuilder;
 use crate::error::FlightError;
@@ -39,8 +40,8 @@ use crate::sql::{
     CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys,
     CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables,
     CommandGetXdbcTypeInfo, CommandPreparedStatementQuery, CommandPreparedStatementUpdate,
-    CommandStatementQuery, CommandStatementUpdate, DoPutPreparedStatementResult, DoPutUpdateResult,
-    ProstMessageExt, SqlInfo,
+    CommandStatementIngest, CommandStatementQuery, CommandStatementUpdate,
+    DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt, SqlInfo,
 };
 use crate::trailers::extract_lazy_trailers;
 use crate::{
@@ -53,10 +54,10 @@ use arrow_ipc::convert::fb_to_schema;
 use arrow_ipc::reader::read_record_batch;
 use arrow_ipc::{root_as_message, MessageHeader};
 use arrow_schema::{ArrowError, Schema, SchemaRef};
-use futures::{stream, TryStreamExt};
+use futures::{stream, Stream, TryStreamExt};
 use prost::Message;
 use tonic::transport::Channel;
-use tonic::{IntoRequest, Streaming};
+use tonic::{IntoRequest, IntoStreamingRequest, Streaming};
 
 /// A FlightSQLServiceClient is an endpoint for retrieving or storing Arrow data
 /// by FlightSQL protocol.
@@ -110,6 +111,11 @@ impl FlightSqlServiceClient<Channel> {
         self.token = None;
     }
 
+    /// Share the bearer token with potentially different `DoGet` clients
+    pub fn token(&self) -> Option<&String> {
+        self.token.as_ref()
+    }
+
     /// Set header value.
     pub fn set_header(&mut self, key: impl Into<String>, value: impl Into<String>) {
         let key: String = key.into();
@@ -227,6 +233,52 @@ impl FlightSqlServiceClient<Channel> {
         Ok(result.record_count)
     }
 
+    /// Execute a bulk ingest on the server and return the number of records added
+    pub async fn execute_ingest<S>(
+        &mut self,
+        command: CommandStatementIngest,
+        stream: S,
+    ) -> Result<i64, ArrowError>
+    where
+        S: Stream<Item = crate::error::Result<RecordBatch>> + Send + 'static,
+    {
+        let (sender, receiver) = futures::channel::oneshot::channel();
+
+        let descriptor = FlightDescriptor::new_cmd(command.as_any().encode_to_vec());
+        let flight_data = FlightDataEncoderBuilder::new()
+            .with_flight_descriptor(Some(descriptor))
+            .build(stream);
+
+        // Intercept client errors and send them to the one shot channel above
+        let flight_data = Box::pin(flight_data);
+        let flight_data: FallibleRequestStream<FlightData, FlightError> =
+            FallibleRequestStream::new(sender, flight_data);
+
+        let req = self.set_request_headers(flight_data.into_streaming_request())?;
+        let mut result = self
+            .flight_client
+            .do_put(req)
+            .await
+            .map_err(status_to_arrow_error)?
+            .into_inner();
+
+        // check if the there were any errors in the input stream provided note
+        // if receiver.await fails, it means the sender was dropped and there is
+        // no message to return.
+        if let Ok(msg) = receiver.await {
+            return Err(ArrowError::ExternalError(Box::new(msg)));
+        }
+
+        let result = result
+            .message()
+            .await
+            .map_err(status_to_arrow_error)?
+            .unwrap();
+        let any = Any::decode(&*result.app_metadata).map_err(decode_error_to_arrow_error)?;
+        let result: DoPutUpdateResult = any.unpack()?.unwrap();
+        Ok(result.record_count)
+    }
+
     /// Request a list of catalogs as tabular FlightInfo results
     pub async fn get_catalogs(&mut self) -> Result<FlightInfo, ArrowError> {
         self.get_flight_info_for_command(CommandGetCatalogs {})
diff --git a/arrow-flight/src/sql/mod.rs b/arrow-flight/src/sql/mod.rs
index 61eb67b69..453f608d3 100644
--- a/arrow-flight/src/sql/mod.rs
+++ b/arrow-flight/src/sql/mod.rs
@@ -50,6 +50,10 @@ mod gen {
 }
 
 pub use gen::action_end_transaction_request::EndTransaction;
+pub use gen::command_statement_ingest::table_definition_options::{
+    TableExistsOption, TableNotExistOption,
+};
+pub use gen::command_statement_ingest::TableDefinitionOptions;
 pub use gen::ActionBeginSavepointRequest;
 pub use gen::ActionBeginSavepointResult;
 pub use gen::ActionBeginTransactionRequest;
@@ -74,6 +78,7 @@ pub use gen::CommandGetTables;
 pub use gen::CommandGetXdbcTypeInfo;
 pub use gen::CommandPreparedStatementQuery;
 pub use gen::CommandPreparedStatementUpdate;
+pub use gen::CommandStatementIngest;
 pub use gen::CommandStatementQuery;
 pub use gen::CommandStatementSubstraitPlan;
 pub use gen::CommandStatementUpdate;
@@ -250,11 +255,12 @@ prost_message_ext!(
     CommandGetXdbcTypeInfo,
     CommandPreparedStatementQuery,
     CommandPreparedStatementUpdate,
+    CommandStatementIngest,
     CommandStatementQuery,
     CommandStatementSubstraitPlan,
     CommandStatementUpdate,
-    DoPutUpdateResult,
     DoPutPreparedStatementResult,
+    DoPutUpdateResult,
     TicketStatementQuery,
 );
 
diff --git a/arrow-flight/src/sql/server.rs b/arrow-flight/src/sql/server.rs
index b47691c7d..e348367a9 100644
--- a/arrow-flight/src/sql/server.rs
+++ b/arrow-flight/src/sql/server.rs
@@ -32,9 +32,9 @@ use super::{
     CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys,
     CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables,
     CommandGetXdbcTypeInfo, CommandPreparedStatementQuery, CommandPreparedStatementUpdate,
-    CommandStatementQuery, CommandStatementSubstraitPlan, CommandStatementUpdate,
-    DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt, SqlInfo,
-    TicketStatementQuery,
+    CommandStatementIngest, CommandStatementQuery, CommandStatementSubstraitPlan,
+    CommandStatementUpdate, DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt,
+    SqlInfo, TicketStatementQuery,
 };
 use crate::{
     flight_service_server::FlightService, gen::PollInfo, Action, ActionType, Criteria, Empty,
@@ -397,6 +397,17 @@ pub trait FlightSqlService: Sync + Send + Sized + 'static {
         ))
     }
 
+    /// Execute a bulk ingestion.
+    async fn do_put_statement_ingest(
+        &self,
+        _ticket: CommandStatementIngest,
+        _request: Request<PeekableFlightDataStream>,
+    ) -> Result<i64, Status> {
+        Err(Status::unimplemented(
+            "do_put_statement_ingest has no default implementation",
+        ))
+    }
+
     /// Bind parameters to given prepared statement.
     ///
     /// Returns an opaque handle that the client should pass
@@ -713,6 +724,14 @@ where
                 })]);
                 Ok(Response::new(Box::pin(output)))
             }
+            Command::CommandStatementIngest(command) => {
+                let record_count = self.do_put_statement_ingest(command, request).await?;
+                let result = DoPutUpdateResult { record_count };
+                let output = futures::stream::iter(vec![Ok(PutResult {
+                    app_metadata: result.as_any().encode_to_vec().into(),
+                })]);
+                Ok(Response::new(Box::pin(output)))
+            }
             Command::CommandPreparedStatementQuery(command) => {
                 let result = self
                     .do_put_prepared_statement_query(command, request)
diff --git a/object_store/CONTRIBUTING.md b/object_store/CONTRIBUTING.md
index 4b0ef1f0e..5444ec743 100644
--- a/object_store/CONTRIBUTING.md
+++ b/object_store/CONTRIBUTING.md
@@ -101,6 +101,54 @@ export AWS_SERVER_SIDE_ENCRYPTION=aws:kms:dsse
 cargo test --features aws
 ```
 
+#### SSE-C Encryption tests
+
+Unfortunately, localstack does not support SSE-C encryption (https://github.com/localstack/localstack/issues/11356).
+
+We will use [MinIO](https://min.io/docs/minio/container/operations/server-side-encryption.html) to test SSE-C encryption.
+
+First, create a self-signed certificate to enable HTTPS for MinIO, as SSE-C requires HTTPS.
+
+```shell
+mkdir ~/certs
+cd ~/certs
+openssl genpkey -algorithm RSA -out private.key
+openssl req -new -key private.key -out request.csr -subj "/C=US/ST=State/L=City/O=Organization/OU=Unit/CN=example.com/emailAddress=email@example.com"
+openssl x509 -req -days 365 -in request.csr -signkey private.key -out public.crt
+rm request.csr
+```
+
+Second, start MinIO with the self-signed certificate.
+
+```shell
+docker run -d \
+  -p 9000:9000 \
+  --name minio \
+  -v ${HOME}/certs:/root/.minio/certs \
+  -e "MINIO_ROOT_USER=minio" \
+  -e "MINIO_ROOT_PASSWORD=minio123" \
+  minio/minio server /data
+```
+
+Create a test bucket.
+
+```shell
+export AWS_BUCKET_NAME=test-bucket
+export AWS_ACCESS_KEY_ID=minio
+export AWS_SECRET_ACCESS_KEY=minio123
+export AWS_ENDPOINT=https://localhost:9000
+aws s3 mb s3://test-bucket --endpoint-url=https://localhost:9000 --no-verify-ssl
+```
+
+Run the tests. The real test is `test_s3_ssec_encryption_with_minio()`
+
+```shell
+export TEST_S3_SSEC_ENCRYPTION=1
+cargo test --features aws --package object_store --lib aws::tests::test_s3_ssec_encryption_with_minio -- --exact --nocapture
+```
+
+
+
 ### Azure
 
 To test the Azure integration
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index ffef3fb33..574345c38 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -26,7 +26,10 @@ use crate::aws::{
 use crate::client::TokenCredentialProvider;
 use crate::config::ConfigValue;
 use crate::{ClientConfigKey, ClientOptions, Result, RetryConfig, StaticCredentialProvider};
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use itertools::Itertools;
+use md5::{Digest, Md5};
 use reqwest::header::{HeaderMap, HeaderValue};
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
@@ -73,7 +76,7 @@ enum Error {
     #[snafu(display("Invalid Zone suffix for bucket '{bucket}'"))]
     ZoneSuffix { bucket: String },
 
-    #[snafu(display("Invalid encryption type: {}. Valid values are \"AES256\", \"sse:kms\", and \"sse:kms:dsse\".", passed))]
+    #[snafu(display("Invalid encryption type: {}. Valid values are \"AES256\", \"sse:kms\", \"sse:kms:dsse\" and \"sse-c\".", passed))]
     InvalidEncryptionType { passed: String },
 
     #[snafu(display(
@@ -166,6 +169,8 @@ pub struct AmazonS3Builder {
     encryption_type: Option<ConfigValue<S3EncryptionType>>,
     encryption_kms_key_id: Option<String>,
     encryption_bucket_key_enabled: Option<ConfigValue<bool>>,
+    /// base64-encoded 256-bit customer encryption key for SSE-C.
+    encryption_customer_key_base64: Option<String>,
 }
 
 /// Configuration keys for [`AmazonS3Builder`]
@@ -394,6 +399,9 @@ impl FromStr for AmazonS3ConfigKey {
             "aws_sse_bucket_key_enabled" => {
                 Ok(Self::Encryption(S3EncryptionConfigKey::BucketKeyEnabled))
             }
+            "aws_sse_customer_key_base64" => Ok(Self::Encryption(
+                S3EncryptionConfigKey::CustomerEncryptionKey,
+            )),
             _ => match s.parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
@@ -511,6 +519,9 @@ impl AmazonS3Builder {
                 S3EncryptionConfigKey::BucketKeyEnabled => {
                     self.encryption_bucket_key_enabled = Some(ConfigValue::Deferred(value.into()))
                 }
+                S3EncryptionConfigKey::CustomerEncryptionKey => {
+                    self.encryption_customer_key_base64 = Some(value.into())
+                }
             },
         };
         self
@@ -566,6 +577,9 @@ impl AmazonS3Builder {
                     .encryption_bucket_key_enabled
                     .as_ref()
                     .map(ToString::to_string),
+                S3EncryptionConfigKey::CustomerEncryptionKey => {
+                    self.encryption_customer_key_base64.clone()
+                }
             },
         }
     }
@@ -813,6 +827,14 @@ impl AmazonS3Builder {
         self
     }
 
+    /// Use SSE-C for server side encryption.
+    /// Must pass the *base64-encoded* 256-bit customer encryption key.
+    pub fn with_ssec_encryption(mut self, customer_key_base64: impl Into<String>) -> Self {
+        self.encryption_type = Some(ConfigValue::Parsed(S3EncryptionType::SseC));
+        self.encryption_customer_key_base64 = customer_key_base64.into().into();
+        self
+    }
+
     /// Set whether to enable bucket key for server side encryption. This overrides
     /// the bucket default setting for bucket keys.
     ///
@@ -953,6 +975,7 @@ impl AmazonS3Builder {
                 self.encryption_bucket_key_enabled
                     .map(|val| val.get())
                     .transpose()?,
+                self.encryption_customer_key_base64,
             )?
         } else {
             S3EncryptionHeaders::default()
@@ -994,15 +1017,14 @@ fn parse_bucket_az(bucket: &str) -> Option<&str> {
 /// These options are used to configure server-side encryption for S3 objects.
 /// To configure them, pass them to [`AmazonS3Builder::with_config`].
 ///
-/// Both [SSE-KMS] and [DSSE-KMS] are supported. [SSE-C] is not yet supported.
-///
+/// [SSE-S3]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html
 /// [SSE-KMS]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingKMSEncryption.html
 /// [DSSE-KMS]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingDSSEncryption.html
 /// [SSE-C]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/ServerSideEncryptionCustomerKeys.html
 #[derive(PartialEq, Eq, Hash, Clone, Debug, Copy, Serialize, Deserialize)]
 #[non_exhaustive]
 pub enum S3EncryptionConfigKey {
-    /// Type of encryption to use. If set, must be one of "AES256", "aws:kms", or "aws:kms:dsse".
+    /// Type of encryption to use. If set, must be one of "AES256" (SSE-S3), "aws:kms" (SSE-KMS), "aws:kms:dsse" (DSSE-KMS) or "sse-c".
     ServerSideEncryption,
     /// The KMS key ID to use for server-side encryption. If set, ServerSideEncryption
     /// must be "aws:kms" or "aws:kms:dsse".
@@ -1010,6 +1032,10 @@ pub enum S3EncryptionConfigKey {
     /// If set to true, will use the bucket's default KMS key for server-side encryption.
     /// If set to false, will disable the use of the bucket's default KMS key for server-side encryption.
     BucketKeyEnabled,
+
+    /// The base64 encoded, 256-bit customer encryption key to use for server-side encryption.
+    /// If set, ServerSideEncryption must be "sse-c".
+    CustomerEncryptionKey,
 }
 
 impl AsRef<str> for S3EncryptionConfigKey {
@@ -1018,6 +1044,7 @@ impl AsRef<str> for S3EncryptionConfigKey {
             Self::ServerSideEncryption => "aws_server_side_encryption",
             Self::KmsKeyId => "aws_sse_kms_key_id",
             Self::BucketKeyEnabled => "aws_sse_bucket_key_enabled",
+            Self::CustomerEncryptionKey => "aws_sse_customer_key_base64",
         }
     }
 }
@@ -1027,6 +1054,7 @@ enum S3EncryptionType {
     S3,
     SseKms,
     DsseKms,
+    SseC,
 }
 
 impl crate::config::Parse for S3EncryptionType {
@@ -1035,6 +1063,7 @@ impl crate::config::Parse for S3EncryptionType {
             "AES256" => Ok(Self::S3),
             "aws:kms" => Ok(Self::SseKms),
             "aws:kms:dsse" => Ok(Self::DsseKms),
+            "sse-c" => Ok(Self::SseC),
             _ => Err(Error::InvalidEncryptionType { passed: s.into() }.into()),
         }
     }
@@ -1046,6 +1075,7 @@ impl From<&S3EncryptionType> for &'static str {
             S3EncryptionType::S3 => "AES256",
             S3EncryptionType::SseKms => "aws:kms",
             S3EncryptionType::DsseKms => "aws:kms:dsse",
+            S3EncryptionType::SseC => "sse-c",
         }
     }
 }
@@ -1062,37 +1092,87 @@ impl std::fmt::Display for S3EncryptionType {
 /// Whether these headers are sent depends on both the kind of encryption set
 /// and the kind of request being made.
 #[derive(Default, Clone, Debug)]
-pub struct S3EncryptionHeaders(HeaderMap);
+pub(super) struct S3EncryptionHeaders(pub HeaderMap);
 
 impl S3EncryptionHeaders {
     fn try_new(
         encryption_type: &S3EncryptionType,
-        key_id: Option<String>,
+        encryption_kms_key_id: Option<String>,
         bucket_key_enabled: Option<bool>,
+        encryption_customer_key_base64: Option<String>,
     ) -> Result<Self> {
         let mut headers = HeaderMap::new();
-        // Note: if we later add support for SSE-C, we should be sure to use
-        // HeaderValue::set_sensitive to prevent the key from being logged.
-        headers.insert(
-            "x-amz-server-side-encryption",
-            HeaderValue::from_static(encryption_type.into()),
-        );
-        if let Some(key_id) = key_id {
-            headers.insert(
-                "x-amz-server-side-encryption-aws-kms-key-id",
-                key_id
-                    .try_into()
-                    .map_err(|err| Error::InvalidEncryptionHeader {
-                        header: "kms-key-id",
-                        source: Box::new(err),
-                    })?,
-            );
-        }
-        if let Some(bucket_key_enabled) = bucket_key_enabled {
-            headers.insert(
-                "x-amz-server-side-encryption-bucket-key-enabled",
-                HeaderValue::from_static(if bucket_key_enabled { "true" } else { "false" }),
-            );
+        match encryption_type {
+            S3EncryptionType::S3 | S3EncryptionType::SseKms | S3EncryptionType::DsseKms => {
+                headers.insert(
+                    "x-amz-server-side-encryption",
+                    HeaderValue::from_static(encryption_type.into()),
+                );
+                if let Some(key_id) = encryption_kms_key_id {
+                    headers.insert(
+                        "x-amz-server-side-encryption-aws-kms-key-id",
+                        key_id
+                            .try_into()
+                            .map_err(|err| Error::InvalidEncryptionHeader {
+                                header: "kms-key-id",
+                                source: Box::new(err),
+                            })?,
+                    );
+                }
+                if let Some(bucket_key_enabled) = bucket_key_enabled {
+                    headers.insert(
+                        "x-amz-server-side-encryption-bucket-key-enabled",
+                        HeaderValue::from_static(if bucket_key_enabled { "true" } else { "false" }),
+                    );
+                }
+            }
+            S3EncryptionType::SseC => {
+                headers.insert(
+                    "x-amz-server-side-encryption-customer-algorithm",
+                    HeaderValue::from_static("AES256"),
+                );
+                if let Some(key) = encryption_customer_key_base64 {
+                    let mut header_value: HeaderValue =
+                        key.clone()
+                            .try_into()
+                            .map_err(|err| Error::InvalidEncryptionHeader {
+                                header: "x-amz-server-side-encryption-customer-key",
+                                source: Box::new(err),
+                            })?;
+                    header_value.set_sensitive(true);
+                    headers.insert("x-amz-server-side-encryption-customer-key", header_value);
+
+                    let decoded_key = BASE64_STANDARD.decode(key.as_bytes()).map_err(|err| {
+                        Error::InvalidEncryptionHeader {
+                            header: "x-amz-server-side-encryption-customer-key",
+                            source: Box::new(err),
+                        }
+                    })?;
+                    let mut hasher = Md5::new();
+                    hasher.update(decoded_key);
+                    let md5 = BASE64_STANDARD.encode(hasher.finalize());
+                    let mut md5_header_value: HeaderValue =
+                        md5.try_into()
+                            .map_err(|err| Error::InvalidEncryptionHeader {
+                                header: "x-amz-server-side-encryption-customer-key-MD5",
+                                source: Box::new(err),
+                            })?;
+                    md5_header_value.set_sensitive(true);
+                    headers.insert(
+                        "x-amz-server-side-encryption-customer-key-MD5",
+                        md5_header_value,
+                    );
+                } else {
+                    return Err(Error::InvalidEncryptionHeader {
+                        header: "x-amz-server-side-encryption-customer-key",
+                        source: Box::new(std::io::Error::new(
+                            std::io::ErrorKind::InvalidInput,
+                            "Missing customer key",
+                        )),
+                    }
+                    .into());
+                }
+            }
         }
         Ok(Self(headers))
     }
@@ -1162,7 +1242,11 @@ mod tests {
             .with_config(AmazonS3ConfigKey::UnsignedPayload, "true")
             .with_config("aws_server_side_encryption".parse().unwrap(), "AES256")
             .with_config("aws_sse_kms_key_id".parse().unwrap(), "some_key_id")
-            .with_config("aws_sse_bucket_key_enabled".parse().unwrap(), "true");
+            .with_config("aws_sse_bucket_key_enabled".parse().unwrap(), "true")
+            .with_config(
+                "aws_sse_customer_key_base64".parse().unwrap(),
+                "some_customer_key",
+            );
 
         assert_eq!(
             builder
@@ -1216,6 +1300,12 @@ mod tests {
                 .unwrap(),
             "true"
         );
+        assert_eq!(
+            builder
+                .get_config_value(&"aws_sse_customer_key_base64".parse().unwrap())
+                .unwrap(),
+            "some_customer_key"
+        );
     }
 
     #[test]
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index ab4da86f5..007e27108 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -181,7 +181,7 @@ pub struct S3Config {
     pub checksum: Option<Checksum>,
     pub copy_if_not_exists: Option<S3CopyIfNotExists>,
     pub conditional_put: Option<S3ConditionalPut>,
-    pub encryption_headers: S3EncryptionHeaders,
+    pub(super) encryption_headers: S3EncryptionHeaders,
 }
 
 impl S3Config {
@@ -522,10 +522,47 @@ impl S3Client {
     /// Make an S3 Copy request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html>
     pub fn copy_request<'a>(&'a self, from: &Path, to: &'a Path) -> Request<'a> {
         let source = format!("{}/{}", self.config.bucket, encode_path(from));
+
+        let mut copy_source_encryption_headers = HeaderMap::new();
+        if let Some(customer_algorithm) = self
+            .config
+            .encryption_headers
+            .0
+            .get("x-amz-server-side-encryption-customer-algorithm")
+        {
+            copy_source_encryption_headers.insert(
+                "x-amz-copy-source-server-side-encryption-customer-algorithm",
+                customer_algorithm.clone(),
+            );
+        }
+        if let Some(customer_key) = self
+            .config
+            .encryption_headers
+            .0
+            .get("x-amz-server-side-encryption-customer-key")
+        {
+            copy_source_encryption_headers.insert(
+                "x-amz-copy-source-server-side-encryption-customer-key",
+                customer_key.clone(),
+            );
+        }
+        if let Some(customer_key_md5) = self
+            .config
+            .encryption_headers
+            .0
+            .get("x-amz-server-side-encryption-customer-key-MD5")
+        {
+            copy_source_encryption_headers.insert(
+                "x-amz-copy-source-server-side-encryption-customer-key-MD5",
+                customer_key_md5.clone(),
+            );
+        }
+
         self.request(Method::PUT, to)
             .idempotent(true)
             .header(&COPY_SOURCE_HEADER, &source)
             .headers(self.config.encryption_headers.clone().into())
+            .headers(copy_source_encryption_headers)
             .with_session_creds(false)
     }
 
@@ -562,13 +599,21 @@ impl S3Client {
     ) -> Result<PartId> {
         let part = (part_idx + 1).to_string();
 
-        let response = self
+        let mut request = self
             .request(Method::PUT, path)
             .with_payload(data)
             .query(&[("partNumber", &part), ("uploadId", upload_id)])
-            .idempotent(true)
-            .send()
-            .await?;
+            .idempotent(true);
+        if self
+            .config
+            .encryption_headers
+            .0
+            .contains_key("x-amz-server-side-encryption-customer-algorithm")
+        {
+            // If SSE-C is used, we must include the encryption headers in every upload request.
+            request = request.with_encryption_headers();
+        }
+        let response = request.send().await?;
 
         let content_id = get_etag(response.headers()).context(MetadataSnafu)?;
         Ok(PartId { content_id })
@@ -660,6 +705,7 @@ impl GetClient for S3Client {
         };
 
         let mut builder = self.client.request(method, url);
+        builder = builder.headers(self.config.encryption_headers.clone().into());
 
         if let Some(v) = &options.version {
             builder = builder.query(&[("versionId", v)])
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index f5204a536..4a773e7a1 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -60,7 +60,7 @@ mod dynamo;
 mod precondition;
 mod resolve;
 
-pub use builder::{AmazonS3Builder, AmazonS3ConfigKey, S3EncryptionHeaders};
+pub use builder::{AmazonS3Builder, AmazonS3ConfigKey};
 pub use checksum::Checksum;
 pub use dynamo::DynamoCommit;
 pub use precondition::{S3ConditionalPut, S3CopyIfNotExists};
@@ -412,6 +412,9 @@ mod tests {
     use crate::client::get::GetClient;
     use crate::integration::*;
     use crate::tests::*;
+    use crate::ClientOptions;
+    use base64::prelude::BASE64_STANDARD;
+    use base64::Engine;
     use hyper::HeaderMap;
 
     const NON_EXISTENT_NAME: &str = "nonexistentname";
@@ -605,4 +608,67 @@ mod tests {
             store.delete(location).await.unwrap();
         }
     }
+
+    /// See CONTRIBUTING.md for the MinIO setup for this test.
+    #[tokio::test]
+    async fn test_s3_ssec_encryption_with_minio() {
+        if std::env::var("TEST_S3_SSEC_ENCRYPTION").is_err() {
+            eprintln!("Skipping S3 SSE-C encryption test");
+            return;
+        }
+        eprintln!("Running S3 SSE-C encryption test");
+
+        let customer_key = "1234567890abcdef1234567890abcdef";
+        let expected_md5 = "JMwgiexXqwuPqIPjYFmIZQ==";
+
+        let store = AmazonS3Builder::from_env()
+            .with_ssec_encryption(BASE64_STANDARD.encode(customer_key))
+            .with_client_options(ClientOptions::default().with_allow_invalid_certificates(true))
+            .build()
+            .unwrap();
+
+        let data = PutPayload::from(vec![3u8; 1024]);
+
+        let locations = [
+            Path::from("test-encryption-1"),
+            Path::from("test-encryption-2"),
+            Path::from("test-encryption-3"),
+        ];
+
+        // Test put with sse-c.
+        store.put(&locations[0], data.clone()).await.unwrap();
+
+        // Test copy with sse-c.
+        store.copy(&locations[0], &locations[1]).await.unwrap();
+
+        // Test multipart upload with sse-c.
+        let mut upload = store.put_multipart(&locations[2]).await.unwrap();
+        upload.put_part(data.clone()).await.unwrap();
+        upload.complete().await.unwrap();
+
+        // Test get with sse-c.
+        for location in &locations {
+            let res = store
+                .client
+                .get_request(location, GetOptions::default())
+                .await
+                .unwrap();
+            let headers = res.headers();
+            assert_eq!(
+                headers
+                    .get("x-amz-server-side-encryption-customer-algorithm")
+                    .expect("object is not encrypted with SSE-C"),
+                "AES256"
+            );
+
+            assert_eq!(
+                headers
+                    .get("x-amz-server-side-encryption-customer-key-MD5")
+                    .expect("object is not encrypted with SSE-C"),
+                expected_md5
+            );
+
+            store.delete(location).await.unwrap();
+        }
+    }
 }
diff --git a/object_store/src/client/mod.rs b/object_store/src/client/mod.rs
index 43fd65892..c45833b89 100644
--- a/object_store/src/client/mod.rs
+++ b/object_store/src/client/mod.rs
@@ -167,10 +167,60 @@ impl FromStr for ClientConfigKey {
     }
 }
 
+/// Represents a CA certificate provided by the user.
+///
+/// This is used to configure the client to trust a specific certificate. See
+/// [Self::from_pem] for an example
+#[derive(Debug, Clone)]
+pub struct Certificate(reqwest::tls::Certificate);
+
+impl Certificate {
+    /// Create a `Certificate` from a PEM encoded certificate.
+    ///
+    /// # Example from a PEM file
+    ///
+    /// ```no_run
+    /// # use object_store::Certificate;
+    /// # use std::fs::File;
+    /// # use std::io::Read;
+    /// let mut buf = Vec::new();
+    /// File::open("my_cert.pem").unwrap()
+    ///   .read_to_end(&mut buf).unwrap();
+    /// let cert = Certificate::from_pem(&buf).unwrap();
+    ///
+    /// ```
+    pub fn from_pem(pem: &[u8]) -> Result<Self> {
+        Ok(Self(
+            reqwest::tls::Certificate::from_pem(pem).map_err(map_client_error)?,
+        ))
+    }
+
+    /// Create a collection of `Certificate` from a PEM encoded certificate
+    /// bundle.
+    ///
+    /// Files that contain such collections have extensions such as `.crt`,
+    /// `.cer` and `.pem` files.
+    pub fn from_pem_bundle(pem_bundle: &[u8]) -> Result<Vec<Self>> {
+        Ok(reqwest::tls::Certificate::from_pem_bundle(pem_bundle)
+            .map_err(map_client_error)?
+            .into_iter()
+            .map(Self)
+            .collect())
+    }
+
+    /// Create a `Certificate` from a binary DER encoded certificate.
+    pub fn from_der(der: &[u8]) -> Result<Self> {
+        Ok(Self(
+            reqwest::tls::Certificate::from_der(der).map_err(map_client_error)?,
+        ))
+    }
+}
+
 /// HTTP client configuration for remote object stores
 #[derive(Debug, Clone)]
 pub struct ClientOptions {
     user_agent: Option<ConfigValue<HeaderValue>>,
+    root_certificates: Vec<Certificate>,
     content_type_map: HashMap<String, String>,
     default_content_type: Option<String>,
     default_headers: Option<HeaderMap>,
@@ -201,6 +251,7 @@ impl Default for ClientOptions {
         // we opt for a slightly higher default timeout of 30 seconds
         Self {
             user_agent: None,
+            root_certificates: Default::default(),
             content_type_map: Default::default(),
             default_content_type: None,
             default_headers: None,
@@ -310,6 +361,15 @@ impl ClientOptions {
         self
     }
 
+    /// Add a custom root certificate.
+    ///
+    /// This can be used to connect to a server that has a self-signed
+    /// certificate for example.
+    pub fn with_root_certificate(mut self, certificate: Certificate) -> Self {
+        self.root_certificates.push(certificate);
+        self
+    }
+
     /// Set the default CONTENT_TYPE for uploads
     pub fn with_default_content_type(mut self, mime: impl Into<String>) -> Self {
         self.default_content_type = Some(mime.into());
@@ -541,6 +601,10 @@ impl ClientOptions {
             builder = builder.proxy(proxy);
         }
 
+        for certificate in &self.root_certificates {
+            builder = builder.add_root_certificate(certificate.0.clone());
+        }
+
         if let Some(timeout) = &self.timeout {
             builder = builder.timeout(timeout.get()?)
         }
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 4184d58a0..4b43f0cde 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -526,8 +526,8 @@ mod client;
 
 #[cfg(feature = "cloud")]
 pub use client::{
-    backoff::BackoffConfig, retry::RetryConfig, ClientConfigKey, ClientOptions, CredentialProvider,
-    StaticCredentialProvider,
+    backoff::BackoffConfig, retry::RetryConfig, Certificate, ClientConfigKey, ClientOptions,
+    CredentialProvider, StaticCredentialProvider,
 };
 
 #[cfg(feature = "cloud")]
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index ad740b0a2..b97b2a571 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -44,7 +44,7 @@ arrow-schema = { workspace = true, optional = true }
 arrow-select = { workspace = true, optional = true }
 arrow-ipc = { workspace = true, optional = true }
 # Intentionally not a path dependency as object_store is released separately
-object_store = { version = "0.10.2", default-features = false, optional = true }
+object_store = { version = "0.11.0", default-features = false, optional = true }
 
 bytes = { version = "1.1", default-features = false, features = ["std"] }
 thrift = { version = "0.17", default-features = false }
@@ -82,7 +82,7 @@ serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-object_store = { version = "0.10.2", default-features = false, features = ["azure"] }
+object_store = { version = "0.11.0", default-features = false, features = ["azure"] }
 
 # TODO: temporary to fix parquet wasm build
 # upstream issue: https://github.com/gyscos/zstd-rs/issues/269
diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs
index a12c9d860..02c2f44f6 100644
--- a/parquet/src/basic.rs
+++ b/parquet/src/basic.rs
@@ -358,6 +358,14 @@ pub enum Compression {
     LZ4_RAW,
 }
 
+impl Compression {
+    /// Returns the codec type of this compression setting as a string, without the compression
+    /// level.
+    pub(crate) fn codec_to_string(self) -> String {
+        format!("{:?}", self).split('(').next().unwrap().to_owned()
+    }
+}
+
 fn split_compression_string(str_setting: &str) -> Result<(&str, Option<u32>), ParquetError> {
     let split_setting = str_setting.split_once('(');
 
@@ -1915,6 +1923,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_compression_codec_to_string() {
+        assert_eq!(Compression::UNCOMPRESSED.codec_to_string(), "UNCOMPRESSED");
+        assert_eq!(
+            Compression::ZSTD(ZstdLevel::default()).codec_to_string(),
+            "ZSTD"
+        );
+    }
+
     #[test]
     fn test_display_compression() {
         assert_eq!(Compression::UNCOMPRESSED.to_string(), "UNCOMPRESSED");
diff --git a/parquet/src/column/reader.rs b/parquet/src/column/reader.rs
index b40ca2b78..0c7cbb412 100644
--- a/parquet/src/column/reader.rs
+++ b/parquet/src/column/reader.rs
@@ -240,6 +240,12 @@ where
                     let (mut records_read, levels_read) =
                         reader.read_rep_levels(out, remaining_records, remaining_levels)?;
 
+                    if records_read == 0 && levels_read == 0 {
+                        // The fact that we're still looping implies there must be some levels to read.
+                        return Err(general_err!(
+                            "Insufficient repetition levels read from column"
+                        ));
+                    }
                     if levels_read == remaining_levels && self.has_record_delimiter {
                         // Reached end of page, which implies records_read < remaining_records
                         // as otherwise would have stopped reading before reaching the end
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 8ea287831..606ad462d 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -1842,7 +1842,7 @@ mod tests {
         assert_eq!(metadata.dictionary_page_offset(), Some(0));
         if let Some(stats) = metadata.statistics() {
             assert_eq!(stats.null_count_opt(), Some(0));
-            assert_eq!(stats.distinct_count(), None);
+            assert_eq!(stats.distinct_count_opt(), None);
             if let Statistics::Int32(stats) = stats {
                 assert_eq!(stats.min_opt().unwrap(), &1);
                 assert_eq!(stats.max_opt().unwrap(), &4);
@@ -1968,7 +1968,7 @@ mod tests {
         assert_eq!(metadata.dictionary_page_offset(), Some(0));
         if let Some(stats) = metadata.statistics() {
             assert_eq!(stats.null_count_opt(), Some(0));
-            assert_eq!(stats.distinct_count().unwrap_or(0), 55);
+            assert_eq!(stats.distinct_count_opt().unwrap_or(0), 55);
             if let Statistics::Int32(stats) = stats {
                 assert_eq!(stats.min_opt().unwrap(), &-17);
                 assert_eq!(stats.max_opt().unwrap(), &9000);
@@ -1999,7 +1999,7 @@ mod tests {
         assert_eq!(stats.min_bytes_opt().unwrap(), 1_i32.to_le_bytes());
         assert_eq!(stats.max_bytes_opt().unwrap(), 7_i32.to_le_bytes());
         assert_eq!(stats.null_count_opt(), Some(0));
-        assert!(stats.distinct_count().is_none());
+        assert!(stats.distinct_count_opt().is_none());
 
         drop(write);
 
@@ -2031,7 +2031,7 @@ mod tests {
             7_i32.to_le_bytes()
         );
         assert_eq!(page_statistics.null_count_opt(), Some(0));
-        assert!(page_statistics.distinct_count().is_none());
+        assert!(page_statistics.distinct_count_opt().is_none());
     }
 
     #[test]
@@ -2698,7 +2698,7 @@ mod tests {
 
         if let Some(stats) = r.metadata.statistics() {
             assert_eq!(stats.null_count_opt(), Some(0));
-            assert_eq!(stats.distinct_count(), None);
+            assert_eq!(stats.distinct_count_opt(), None);
             if let Statistics::Int32(stats) = stats {
                 // first page is [1,2,3,4]
                 // second page is [-5,2,4,8]
@@ -2758,7 +2758,7 @@ mod tests {
 
         if let Some(stats) = r.metadata.statistics() {
             assert_eq!(stats.null_count_opt(), Some(0));
-            assert_eq!(stats.distinct_count(), None);
+            assert_eq!(stats.distinct_count_opt(), None);
             if let Statistics::FixedLenByteArray(stats) = stats {
                 let column_index_min_value = &column_index.min_values[0];
                 let column_index_max_value = &column_index.max_values[0];
@@ -2830,7 +2830,7 @@ mod tests {
 
         if let Some(stats) = r.metadata.statistics() {
             assert_eq!(stats.null_count_opt(), Some(0));
-            assert_eq!(stats.distinct_count(), None);
+            assert_eq!(stats.distinct_count_opt(), None);
             if let Statistics::FixedLenByteArray(_stats) = stats {
                 let column_index_min_value = &column_index.min_values[0];
                 let column_index_max_value = &column_index.max_values[0];
@@ -2951,7 +2951,7 @@ mod tests {
 
         let stats = r.metadata.statistics().expect("statistics");
         assert_eq!(stats.null_count_opt(), Some(0));
-        assert_eq!(stats.distinct_count(), None);
+        assert_eq!(stats.distinct_count_opt(), None);
         if let Statistics::ByteArray(_stats) = stats {
             let min_value = _stats.min_opt().unwrap();
             let max_value = _stats.max_opt().unwrap();
@@ -3003,7 +3003,7 @@ mod tests {
 
         let stats = r.metadata.statistics().expect("statistics");
         assert_eq!(stats.null_count_opt(), Some(0));
-        assert_eq!(stats.distinct_count(), None);
+        assert_eq!(stats.distinct_count_opt(), None);
         if let Statistics::FixedLenByteArray(_stats) = stats {
             let min_value = _stats.min_opt().unwrap();
             let max_value = _stats.max_opt().unwrap();
diff --git a/parquet/src/file/statistics.rs b/parquet/src/file/statistics.rs
index 4134685ff..680c75d6b 100644
--- a/parquet/src/file/statistics.rs
+++ b/parquet/src/file/statistics.rs
@@ -257,7 +257,7 @@ pub fn to_thrift(stats: Option<&Statistics>) -> Option<TStatistics> {
         max: None,
         min: None,
         null_count,
-        distinct_count: stats.distinct_count().map(|value| value as i64),
+        distinct_count: stats.distinct_count_opt().map(|value| value as i64),
         max_value: None,
         min_value: None,
         is_max_value_exact: None,
@@ -380,7 +380,14 @@ impl Statistics {
 
     /// Returns optional value of number of distinct values occurring.
     /// When it is `None`, the value should be ignored.
+    #[deprecated(since = "53.0.0", note = "Use `distinct_count_opt` method instead")]
     pub fn distinct_count(&self) -> Option<u64> {
+        self.distinct_count_opt()
+    }
+
+    /// Returns optional value of number of distinct values occurring.
+    /// When it is `None`, the value should be ignored.
+    pub fn distinct_count_opt(&self) -> Option<u64> {
         statistics_enum_func![self, distinct_count]
     }
 
diff --git a/parquet/src/schema/printer.rs b/parquet/src/schema/printer.rs
index 0bbf2af74..32d727427 100644
--- a/parquet/src/schema/printer.rs
+++ b/parquet/src/schema/printer.rs
@@ -130,7 +130,11 @@ fn print_column_chunk_metadata(out: &mut dyn io::Write, cc_metadata: &ColumnChun
     writeln!(out, "file path: {file_path_str}");
     writeln!(out, "file offset: {}", cc_metadata.file_offset());
     writeln!(out, "num of values: {}", cc_metadata.num_values());
-    writeln!(out, "compression: {}", cc_metadata.compression());
+    writeln!(
+        out,
+        "compression: {}",
+        cc_metadata.compression().codec_to_string()
+    );
     writeln!(
         out,
         "total compressed size (in bytes): {}",


Test patch:
diff --git a/arrow-flight/tests/common/fixture.rs b/arrow-flight/tests/common/fixture.rs
index 141879e2a..a666fa5d0 100644
--- a/arrow-flight/tests/common/fixture.rs
+++ b/arrow-flight/tests/common/fixture.rs
@@ -41,6 +41,7 @@ pub struct TestFixture {
 
 impl TestFixture {
     /// create a new test fixture from the server
+    #[allow(dead_code)]
     pub async fn new<T: FlightService>(test_server: FlightServiceServer<T>) -> Self {
         // let OS choose a free port
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
diff --git a/arrow-flight/tests/common/mod.rs b/arrow-flight/tests/common/mod.rs
index 85716e560..c4ac027c5 100644
--- a/arrow-flight/tests/common/mod.rs
+++ b/arrow-flight/tests/common/mod.rs
@@ -18,3 +18,4 @@
 pub mod fixture;
 pub mod server;
 pub mod trailers_layer;
+pub mod utils;
diff --git a/arrow-flight/tests/common/utils.rs b/arrow-flight/tests/common/utils.rs
new file mode 100644
index 000000000..0f70e4b31
--- /dev/null
+++ b/arrow-flight/tests/common/utils.rs
@@ -0,0 +1,118 @@
+// 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.
+
+//! Common utilities for testing flight clients and servers
+
+use std::sync::Arc;
+
+use arrow_array::{
+    types::Int32Type, ArrayRef, BinaryViewArray, DictionaryArray, Float64Array, RecordBatch,
+    StringViewArray, UInt8Array,
+};
+use arrow_schema::{DataType, Field, Schema};
+
+/// Make a primitive batch for testing
+///
+/// Example:
+/// i: 0, 1, None, 3, 4
+/// f: 5.0, 4.0, None, 2.0, 1.0
+#[allow(dead_code)]
+pub fn make_primitive_batch(num_rows: usize) -> RecordBatch {
+    let i: UInt8Array = (0..num_rows)
+        .map(|i| {
+            if i == num_rows / 2 {
+                None
+            } else {
+                Some(i.try_into().unwrap())
+            }
+        })
+        .collect();
+
+    let f: Float64Array = (0..num_rows)
+        .map(|i| {
+            if i == num_rows / 2 {
+                None
+            } else {
+                Some((num_rows - i) as f64)
+            }
+        })
+        .collect();
+
+    RecordBatch::try_from_iter(vec![("i", Arc::new(i) as ArrayRef), ("f", Arc::new(f))]).unwrap()
+}
+
+/// Make a dictionary batch for testing
+///
+/// Example:
+/// a: value0, value1, value2, None, value1, value2
+#[allow(dead_code)]
+pub fn make_dictionary_batch(num_rows: usize) -> RecordBatch {
+    let values: Vec<_> = (0..num_rows)
+        .map(|i| {
+            if i == num_rows / 2 {
+                None
+            } else {
+                // repeat some values for low cardinality
+                let v = i / 3;
+                Some(format!("value{v}"))
+            }
+        })
+        .collect();
+
+    let a: DictionaryArray<Int32Type> = values
+        .iter()
+        .map(|s| s.as_ref().map(|s| s.as_str()))
+        .collect();
+
+    RecordBatch::try_from_iter(vec![("a", Arc::new(a) as ArrayRef)]).unwrap()
+}
+
+#[allow(dead_code)]
+pub fn make_view_batches(num_rows: usize) -> RecordBatch {
+    const LONG_TEST_STRING: &str =
+        "This is a long string to make sure binary view array handles it";
+    let schema = Schema::new(vec![
+        Field::new("field1", DataType::BinaryView, true),
+        Field::new("field2", DataType::Utf8View, true),
+    ]);
+
+    let string_view_values: Vec<Option<&str>> = (0..num_rows)
+        .map(|i| match i % 3 {
+            0 => None,
+            1 => Some("foo"),
+            2 => Some(LONG_TEST_STRING),
+            _ => unreachable!(),
+        })
+        .collect();
+
+    let bin_view_values: Vec<Option<&[u8]>> = (0..num_rows)
+        .map(|i| match i % 3 {
+            0 => None,
+            1 => Some("bar".as_bytes()),
+            2 => Some(LONG_TEST_STRING.as_bytes()),
+            _ => unreachable!(),
+        })
+        .collect();
+
+    let binary_array = BinaryViewArray::from_iter(bin_view_values);
+    let utf8_array = StringViewArray::from_iter(string_view_values);
+    RecordBatch::try_new(
+        Arc::new(schema.clone()),
+        vec![Arc::new(binary_array), Arc::new(utf8_array)],
+    )
+    .unwrap()
+}
diff --git a/arrow-flight/tests/encode_decode.rs b/arrow-flight/tests/encode_decode.rs
index 0185fa77f..cbfae1825 100644
--- a/arrow-flight/tests/encode_decode.rs
+++ b/arrow-flight/tests/encode_decode.rs
@@ -19,11 +19,7 @@
 
 use std::{collections::HashMap, sync::Arc};
 
-use arrow_array::types::Int32Type;
-use arrow_array::{
-    ArrayRef, BinaryViewArray, DictionaryArray, Float64Array, RecordBatch, StringViewArray,
-    UInt8Array,
-};
+use arrow_array::{ArrayRef, RecordBatch};
 use arrow_cast::pretty::pretty_format_batches;
 use arrow_flight::flight_descriptor::DescriptorType;
 use arrow_flight::FlightDescriptor;
@@ -36,6 +32,9 @@ use arrow_schema::{DataType, Field, Fields, Schema, SchemaRef};
 use bytes::Bytes;
 use futures::{StreamExt, TryStreamExt};
 
+mod common;
+use common::utils::{make_dictionary_batch, make_primitive_batch, make_view_batches};
+
 #[tokio::test]
 async fn test_empty() {
     roundtrip(vec![]).await;
@@ -415,95 +414,6 @@ async fn test_mismatched_schema_message() {
     .await;
 }
 
-/// Make a primitive batch for testing
-///
-/// Example:
-/// i: 0, 1, None, 3, 4
-/// f: 5.0, 4.0, None, 2.0, 1.0
-fn make_primitive_batch(num_rows: usize) -> RecordBatch {
-    let i: UInt8Array = (0..num_rows)
-        .map(|i| {
-            if i == num_rows / 2 {
-                None
-            } else {
-                Some(i.try_into().unwrap())
-            }
-        })
-        .collect();
-
-    let f: Float64Array = (0..num_rows)
-        .map(|i| {
-            if i == num_rows / 2 {
-                None
-            } else {
-                Some((num_rows - i) as f64)
-            }
-        })
-        .collect();
-
-    RecordBatch::try_from_iter(vec![("i", Arc::new(i) as ArrayRef), ("f", Arc::new(f))]).unwrap()
-}
-
-/// Make a dictionary batch for testing
-///
-/// Example:
-/// a: value0, value1, value2, None, value1, value2
-fn make_dictionary_batch(num_rows: usize) -> RecordBatch {
-    let values: Vec<_> = (0..num_rows)
-        .map(|i| {
-            if i == num_rows / 2 {
-                None
-            } else {
-                // repeat some values for low cardinality
-                let v = i / 3;
-                Some(format!("value{v}"))
-            }
-        })
-        .collect();
-
-    let a: DictionaryArray<Int32Type> = values
-        .iter()
-        .map(|s| s.as_ref().map(|s| s.as_str()))
-        .collect();
-
-    RecordBatch::try_from_iter(vec![("a", Arc::new(a) as ArrayRef)]).unwrap()
-}
-
-fn make_view_batches(num_rows: usize) -> RecordBatch {
-    const LONG_TEST_STRING: &str =
-        "This is a long string to make sure binary view array handles it";
-    let schema = Schema::new(vec![
-        Field::new("field1", DataType::BinaryView, true),
-        Field::new("field2", DataType::Utf8View, true),
-    ]);
-
-    let string_view_values: Vec<Option<&str>> = (0..num_rows)
-        .map(|i| match i % 3 {
-            0 => None,
-            1 => Some("foo"),
-            2 => Some(LONG_TEST_STRING),
-            _ => unreachable!(),
-        })
-        .collect();
-
-    let bin_view_values: Vec<Option<&[u8]>> = (0..num_rows)
-        .map(|i| match i % 3 {
-            0 => None,
-            1 => Some("bar".as_bytes()),
-            2 => Some(LONG_TEST_STRING.as_bytes()),
-            _ => unreachable!(),
-        })
-        .collect();
-
-    let binary_array = BinaryViewArray::from_iter(bin_view_values);
-    let utf8_array = StringViewArray::from_iter(string_view_values);
-    RecordBatch::try_new(
-        Arc::new(schema.clone()),
-        vec![Arc::new(binary_array), Arc::new(utf8_array)],
-    )
-    .unwrap()
-}
-
 /// Encodes input as a FlightData stream, and then decodes it using
 /// FlightRecordBatchStream and validates the decoded record batches
 /// match the input.
diff --git a/arrow-flight/tests/flight_sql_client.rs b/arrow-flight/tests/flight_sql_client.rs
index 94b768a13..349da062a 100644
--- a/arrow-flight/tests/flight_sql_client.rs
+++ b/arrow-flight/tests/flight_sql_client.rs
@@ -18,14 +18,21 @@
 mod common;
 
 use crate::common::fixture::TestFixture;
+use crate::common::utils::make_primitive_batch;
+
+use arrow_array::RecordBatch;
+use arrow_flight::decode::FlightRecordBatchStream;
+use arrow_flight::error::FlightError;
 use arrow_flight::flight_service_server::FlightServiceServer;
 use arrow_flight::sql::client::FlightSqlServiceClient;
-use arrow_flight::sql::server::FlightSqlService;
+use arrow_flight::sql::server::{FlightSqlService, PeekableFlightDataStream};
 use arrow_flight::sql::{
     ActionBeginTransactionRequest, ActionBeginTransactionResult, ActionEndTransactionRequest,
-    EndTransaction, SqlInfo,
+    CommandStatementIngest, EndTransaction, SqlInfo, TableDefinitionOptions, TableExistsOption,
+    TableNotExistOption,
 };
 use arrow_flight::Action;
+use futures::{StreamExt, TryStreamExt};
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::Mutex;
@@ -34,9 +41,7 @@ use uuid::Uuid;
 
 #[tokio::test]
 pub async fn test_begin_end_transaction() {
-    let test_server = FlightSqlServiceImpl {
-        transactions: Arc::new(Mutex::new(HashMap::new())),
-    };
+    let test_server = FlightSqlServiceImpl::new();
     let fixture = TestFixture::new(test_server.service()).await;
     let channel = fixture.channel().await;
     let mut flight_sql_client = FlightSqlServiceClient::new(channel);
@@ -63,12 +68,83 @@ pub async fn test_begin_end_transaction() {
         .is_err());
 }
 
+#[tokio::test]
+pub async fn test_execute_ingest() {
+    let test_server = FlightSqlServiceImpl::new();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let channel = fixture.channel().await;
+    let mut flight_sql_client = FlightSqlServiceClient::new(channel);
+    let cmd = make_ingest_command();
+    let expected_rows = 10;
+    let batches = vec![
+        make_primitive_batch(5),
+        make_primitive_batch(3),
+        make_primitive_batch(2),
+    ];
+    let actual_rows = flight_sql_client
+        .execute_ingest(cmd, futures::stream::iter(batches.clone()).map(Ok))
+        .await
+        .expect("ingest should succeed");
+    assert_eq!(actual_rows, expected_rows);
+    // make sure the batches made it through to the server
+    let ingested_batches = test_server.ingested_batches.lock().await.clone();
+    assert_eq!(ingested_batches, batches);
+}
+
+#[tokio::test]
+pub async fn test_execute_ingest_error() {
+    let test_server = FlightSqlServiceImpl::new();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let channel = fixture.channel().await;
+    let mut flight_sql_client = FlightSqlServiceClient::new(channel);
+    let cmd = make_ingest_command();
+    // send an error from the client
+    let batches = vec![
+        Ok(make_primitive_batch(5)),
+        Err(FlightError::NotYetImplemented(
+            "Client error message".to_string(),
+        )),
+    ];
+    // make sure the client returns the error from the client
+    let err = flight_sql_client
+        .execute_ingest(cmd, futures::stream::iter(batches))
+        .await
+        .unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "External error: Not yet implemented: Client error message"
+    );
+}
+
+fn make_ingest_command() -> CommandStatementIngest {
+    CommandStatementIngest {
+        table_definition_options: Some(TableDefinitionOptions {
+            if_not_exist: TableNotExistOption::Create.into(),
+            if_exists: TableExistsOption::Fail.into(),
+        }),
+        table: String::from("test"),
+        schema: None,
+        catalog: None,
+        temporary: true,
+        transaction_id: None,
+        options: HashMap::default(),
+    }
+}
+
 #[derive(Clone)]
 pub struct FlightSqlServiceImpl {
     transactions: Arc<Mutex<HashMap<String, ()>>>,
+    ingested_batches: Arc<Mutex<Vec<RecordBatch>>>,
 }
 
 impl FlightSqlServiceImpl {
+    pub fn new() -> Self {
+        Self {
+            transactions: Arc::new(Mutex::new(HashMap::new())),
+            ingested_batches: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+
     /// Return an [`FlightServiceServer`] that can be used with a
     /// [`Server`](tonic::transport::Server)
     pub fn service(&self) -> FlightServiceServer<Self> {
@@ -77,6 +153,12 @@ impl FlightSqlServiceImpl {
     }
 }
 
+impl Default for FlightSqlServiceImpl {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[tonic::async_trait]
 impl FlightSqlService for FlightSqlServiceImpl {
     type FlightService = FlightSqlServiceImpl;
@@ -116,4 +198,19 @@ impl FlightSqlService for FlightSqlServiceImpl {
     }
 
     async fn register_sql_info(&self, _id: i32, _result: &SqlInfo) {}
+
+    async fn do_put_statement_ingest(
+        &self,
+        _ticket: CommandStatementIngest,
+        request: Request<PeekableFlightDataStream>,
+    ) -> Result<i64, Status> {
+        let batches: Vec<RecordBatch> = FlightRecordBatchStream::new_from_flight_data(
+            request.into_inner().map_err(|e| e.into()),
+        )
+        .try_collect()
+        .await?;
+        let affected_rows = batches.iter().map(|batch| batch.num_rows() as i64).sum();
+        *self.ingested_batches.lock().await.as_mut() = batches;
+        Ok(affected_rows)
+    }
 }
diff --git a/parquet-testing b/parquet-testing
index 9b48ff4f9..cb7a96741 160000
--- a/parquet-testing
+++ b/parquet-testing
@@ -1,1 +1,1 @@
-Subproject commit 9b48ff4f94dc5e89592d46a119884dbb88100884
+Subproject commit cb7a9674142c137367bf75a01b79c6e214a73199
diff --git a/parquet/tests/arrow_reader/bad_data.rs b/parquet/tests/arrow_reader/bad_data.rs
new file mode 100644
index 000000000..6e325f119
--- /dev/null
+++ b/parquet/tests/arrow_reader/bad_data.rs
@@ -0,0 +1,136 @@
+// 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.
+
+//! Tests that reading invalid parquet files returns an error
+
+use arrow::util::test_util::parquet_test_data;
+use parquet::arrow::arrow_reader::ArrowReaderBuilder;
+use parquet::errors::ParquetError;
+use std::collections::HashSet;
+use std::path::PathBuf;
+
+static KNOWN_FILES: &[&str] = &[
+    "PARQUET-1481.parquet",
+    "ARROW-GH-41317.parquet",
+    "ARROW-GH-41321.parquet",
+    "ARROW-RS-GH-6229-DICTHEADER.parquet",
+    "ARROW-RS-GH-6229-LEVELS.parquet",
+    "README.md",
+];
+
+/// Returns the path to 'parquet-testing/bad_data'
+fn bad_data_dir() -> PathBuf {
+    // points to parquet-testing/data
+    let parquet_testing_data = parquet_test_data();
+    PathBuf::from(parquet_testing_data)
+        .parent()
+        .expect("was in parquet-testing/data")
+        .join("bad_data")
+}
+
+#[test]
+// Ensure that if we add a new test the files are added to the tests.
+fn test_invalid_files() {
+    let known_files: HashSet<_> = KNOWN_FILES.iter().cloned().collect();
+    let mut seen_files = HashSet::new();
+
+    let files = std::fs::read_dir(bad_data_dir()).unwrap();
+
+    for file in files {
+        let file_name = file
+            .unwrap()
+            .path()
+            .file_name()
+            .unwrap()
+            .to_str()
+            .unwrap()
+            .to_string();
+
+        // If you see this error, please add a test for the new file following the model below
+        assert!(
+            known_files.contains(file_name.as_str()),
+            "Found new file in bad_data, please add test: {file_name}"
+        );
+        seen_files.insert(file_name);
+    }
+    for expected_file in known_files {
+        assert!(
+            seen_files.contains(expected_file),
+            "Expected file not found in bad_data directory: {expected_file}"
+        );
+    }
+}
+
+#[test]
+fn test_parquet_1481() {
+    let err = read_file("PARQUET-1481.parquet").unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "Parquet error: unexpected parquet type: -7"
+    );
+}
+
+#[test]
+#[should_panic(expected = "assertion failed: self.current_value.is_some()")]
+fn test_arrow_gh_41321() {
+    let err = read_file("ARROW-GH-41321.parquet").unwrap_err();
+    assert_eq!(err.to_string(), "TBD (currently panics)");
+}
+
+#[test]
+fn test_arrow_gh_41317() {
+    let err = read_file("ARROW-GH-41317.parquet").unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "External: Parquet argument error: External: bad data"
+    );
+}
+
+#[test]
+fn test_arrow_rs_gh_6229_dict_header() {
+    let err = read_file("ARROW-RS-GH-6229-DICTHEADER.parquet").unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "External: Parquet argument error: EOF: eof decoding byte array"
+    );
+}
+
+#[test]
+fn test_arrow_rs_gh_6229_dict_levels() {
+    let err = read_file("ARROW-RS-GH-6229-LEVELS.parquet").unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "External: Parquet argument error: Parquet error: Insufficient repetition levels read from column"
+    );
+}
+
+/// Reads the file and tries to return the total row count
+/// Returns an error if the file is invalid
+fn read_file(name: &str) -> Result<usize, ParquetError> {
+    let path = bad_data_dir().join(name);
+    println!("Reading file: {:?}", path);
+
+    let file = std::fs::File::open(&path).unwrap();
+    let reader = ArrowReaderBuilder::try_new(file)?.build()?;
+
+    let mut num_rows = 0;
+    for batch in reader {
+        let batch = batch?;
+        num_rows += batch.num_rows();
+    }
+    Ok(num_rows)
+}
diff --git a/parquet/tests/arrow_reader/mod.rs b/parquet/tests/arrow_reader/mod.rs
index 7e979dcf3..cc4c8f3c9 100644
--- a/parquet/tests/arrow_reader/mod.rs
+++ b/parquet/tests/arrow_reader/mod.rs
@@ -35,6 +35,7 @@ use parquet::file::properties::{EnabledStatistics, WriterProperties};
 use std::sync::Arc;
 use tempfile::NamedTempFile;
 
+mod bad_data;
 mod statistics;
 
 // returns a struct array with columns "int32_col", "float32_col" and "float64_col" with the specified values