Repository: apache/arrow-rs
Pull request #6258 (2024-08-15 19:20:09)

Minor: move `FallibleRequestStream` and `FallibleTonicResponseStream` to a module
# Which issue does this PR close?
Follow on to https://github.com/apache/arrow-rs/pull/6201

<!--
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.
-->
N/A

# Rationale for this change
 
While working on https://github.com/apache/arrow-rs/pull/6201 with @djanderson I was looking for  `FallibleRequestStream` and `FallibleTonicResponseStream` but couldn't find them easily



<!--
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.
-->

# What changes are included in this PR?
Let's put them into their own module with some better documentation to make them
more discoverable

<!--
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?
No, these are crate private structs at the moment

<!--
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/.github/actions/assign.mjs b/.github/actions/assign.mjs
new file mode 100644
index 000000000..d08aa21ec
--- /dev/null
+++ b/.github/actions/assign.mjs
@@ -0,0 +1,27 @@
+// 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.
+
+import { Octokit } from "@octokit/action";
+
+const [owner, repo] = process.env.GITHUB_REPOSITORY.split("/");
+
+new Octokit().rest.issues.addAssignees({
+  owner,
+  repo,
+  issue_number: process.env.ISSUE_NUMBER,
+  assignees: [process.env.ASSIGNEE],
+});
diff --git a/.github/workflows/take.yml b/.github/workflows/take.yml
index 86dc190ad..4985e151d 100644
--- a/.github/workflows/take.yml
+++ b/.github/workflows/take.yml
@@ -25,17 +25,16 @@ permissions:
 
 jobs:
   issue_assign:
-    runs-on: ubuntu-latest
     if: (!github.event.issue.pull_request) && github.event.comment.body == 'take'
-    concurrency:
-      group: ${{ github.actor }}-issue-assign
+    runs-on: ubuntu-latest
     steps:
-      - run: |
-          CODE=$(curl -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" -LI https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.issue.number }}/assignees/${{ github.event.comment.user.login }} -o /dev/null -w '%{http_code}\n' -s)
-          if [ "$CODE" -eq "204" ]
-          then
-            echo "Assigning issue ${{ github.event.issue.number }} to ${{ github.event.comment.user.login }}"
-            curl -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" -d '{"assignees": ["${{ github.event.comment.user.login }}"]}' https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.issue.number }}/assignees
-          else
-            echo "Cannot assign issue ${{ github.event.issue.number }} to ${{ github.event.comment.user.login }}"
-          fi
\ No newline at end of file
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
+      - run: npm install @octokit/action
+      - run: node .github/actions/assign.mjs
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          ISSUE_NUMBER: ${{ github.event.issue.number }}
+          ASSIGNEE: ${{ github.event.comment.user.login }}
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/src/client.rs b/arrow-flight/src/client.rs
index af3c8fba3..97d9899a9 100644
--- a/arrow-flight/src/client.rs
+++ b/arrow-flight/src/client.rs
@@ -15,8 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::{pin::Pin, task::Poll};
-
 use crate::{
     decode::FlightRecordBatchStream,
     flight_service_client::FlightServiceClient,
@@ -28,16 +26,15 @@ use crate::{
 use arrow_schema::Schema;
 use bytes::Bytes;
 use futures::{
-    channel::oneshot::{Receiver, Sender},
     future::ready,
-    ready,
     stream::{self, BoxStream},
-    FutureExt, Stream, StreamExt, TryStreamExt,
+    Stream, StreamExt, TryStreamExt,
 };
 use prost::Message;
 use tonic::{metadata::MetadataMap, transport::Channel};
 
 use crate::error::{FlightError, Result};
+use crate::streams::{FallibleRequestStream, FallibleTonicResponseStream};
 
 /// A "Mid level" [Apache Arrow Flight](https://arrow.apache.org/docs/format/Flight.html) client.
 ///
@@ -674,103 +671,3 @@ impl FlightClient {
         request
     }
 }
-
-/// Wrapper around fallible stream such that when
-/// 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.
-pub(crate) struct FallibleRequestStream<T, E> {
-    /// sender to notify error
-    sender: Option<Sender<E>>,
-    /// fallible stream
-    fallible_stream: Pin<Box<dyn Stream<Item = std::result::Result<T, E>> + Send + 'static>>,
-}
-
-impl<T, E> FallibleRequestStream<T, E> {
-    pub(crate) fn new(
-        sender: Sender<E>,
-        fallible_stream: Pin<Box<dyn Stream<Item = std::result::Result<T, E>> + Send + 'static>>,
-    ) -> Self {
-        Self {
-            sender: Some(sender),
-            fallible_stream,
-        }
-    }
-}
-
-impl<T, E> Stream for FallibleRequestStream<T, E> {
-    type Item = T;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Option<Self::Item>> {
-        let pinned = self.get_mut();
-        let mut request_streams = pinned.fallible_stream.as_mut();
-        match ready!(request_streams.poll_next_unpin(cx)) {
-            Some(Ok(data)) => Poll::Ready(Some(data)),
-            Some(Err(e)) => {
-                // in theory this should only ever be called once
-                // as this stream should not be polled again after returning
-                // None, however we still check for None to be safe
-                if let Some(sender) = pinned.sender.take() {
-                    // an error means the other end of the channel is not around
-                    // to receive the error, so ignore it
-                    let _ = sender.send(e);
-                }
-                Poll::Ready(None)
-            }
-            None => Poll::Ready(None),
-        }
-    }
-}
-
-/// Wrapper for a tonic response stream that can produce a tonic
-/// error. This is tied to a oneshot receiver which can be notified
-/// of other errors. When it receives an error through receiver
-/// end, it prioritises that error to be sent back. See `do_put` or
-/// `do_exchange` for it's uses
-struct FallibleTonicResponseStream<T> {
-    /// Receiver for FlightError
-    receiver: Receiver<FlightError>,
-    /// Tonic response stream
-    response_stream:
-        Pin<Box<dyn Stream<Item = std::result::Result<T, tonic::Status>> + Send + 'static>>,
-}
-
-impl<T> FallibleTonicResponseStream<T> {
-    fn new(
-        receiver: Receiver<FlightError>,
-        response_stream: Pin<
-            Box<dyn Stream<Item = std::result::Result<T, tonic::Status>> + Send + 'static>,
-        >,
-    ) -> Self {
-        Self {
-            receiver,
-            response_stream,
-        }
-    }
-}
-
-impl<T> Stream for FallibleTonicResponseStream<T> {
-    type Item = Result<T>;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
-        let pinned = self.get_mut();
-        let receiver = &mut pinned.receiver;
-        // Prioritise sending the error that's been notified over
-        // polling the response_stream
-        if let Poll::Ready(Ok(err)) = receiver.poll_unpin(cx) {
-            return Poll::Ready(Some(Err(err)));
-        };
-
-        match ready!(pinned.response_stream.poll_next_unpin(cx)) {
-            Some(Ok(res)) => Poll::Ready(Some(Ok(res))),
-            Some(Err(status)) => Poll::Ready(Some(Err(FlightError::Tonic(status)))),
-            None => Poll::Ready(None),
-        }
-    }
-}
diff --git a/arrow-flight/src/lib.rs b/arrow-flight/src/lib.rs
index 8fa61b1d5..1180264e5 100644
--- a/arrow-flight/src/lib.rs
+++ b/arrow-flight/src/lib.rs
@@ -120,6 +120,7 @@ pub mod utils;
 
 #[cfg(feature = "flight-sql-experimental")]
 pub mod sql;
+mod streams;
 
 use flight_descriptor::DescriptorType;
 
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index 9f9963c92..ef52aa27e 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -24,7 +24,6 @@ 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;
@@ -43,6 +42,7 @@ use crate::sql::{
     CommandStatementIngest, CommandStatementQuery, CommandStatementUpdate,
     DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt, SqlInfo,
 };
+use crate::streams::FallibleRequestStream;
 use crate::trailers::extract_lazy_trailers;
 use crate::{
     Action, FlightData, FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse,
@@ -111,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();
diff --git a/arrow-flight/src/streams.rs b/arrow-flight/src/streams.rs
new file mode 100644
index 000000000..e532a80e1
--- /dev/null
+++ b/arrow-flight/src/streams.rs
@@ -0,0 +1,134 @@
+// 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.
+
+//! [`FallibleRequestStream`] and [`FallibleTonicResponseStream`] adapters
+
+use crate::error::FlightError;
+use futures::{
+    channel::oneshot::{Receiver, Sender},
+    FutureExt, Stream, StreamExt,
+};
+use std::pin::Pin;
+use std::task::{ready, Poll};
+
+/// Wrapper around a fallible stream (one that returns errors) that makes it infallible.
+///
+/// Any errors encountered in the stream are ignored are sent to the provided
+/// oneshot sender.
+///
+/// This can be used to accept a stream of `Result<_>` from a client API and send
+/// them to the remote server that wants only the successful results.
+pub(crate) struct FallibleRequestStream<T, E> {
+    /// sender to notify error
+    sender: Option<Sender<E>>,
+    /// fallible stream
+    fallible_stream: Pin<Box<dyn Stream<Item = std::result::Result<T, E>> + Send + 'static>>,
+}
+
+impl<T, E> FallibleRequestStream<T, E> {
+    pub(crate) fn new(
+        sender: Sender<E>,
+        fallible_stream: Pin<Box<dyn Stream<Item = std::result::Result<T, E>> + Send + 'static>>,
+    ) -> Self {
+        Self {
+            sender: Some(sender),
+            fallible_stream,
+        }
+    }
+}
+
+impl<T, E> Stream for FallibleRequestStream<T, E> {
+    type Item = T;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let pinned = self.get_mut();
+        let mut request_streams = pinned.fallible_stream.as_mut();
+        match ready!(request_streams.poll_next_unpin(cx)) {
+            Some(Ok(data)) => Poll::Ready(Some(data)),
+            Some(Err(e)) => {
+                // in theory this should only ever be called once
+                // as this stream should not be polled again after returning
+                // None, however we still check for None to be safe
+                if let Some(sender) = pinned.sender.take() {
+                    // an error means the other end of the channel is not around
+                    // to receive the error, so ignore it
+                    let _ = sender.send(e);
+                }
+                Poll::Ready(None)
+            }
+            None => Poll::Ready(None),
+        }
+    }
+}
+
+/// Wrapper for a tonic response stream that maps errors to `FlightError` and
+/// returns errors from a oneshot channel into the stream.
+///
+/// The user of this stream can inject an error into the response stream using
+/// the one shot receiver. This is used to propagate errors in
+/// [`FlightClient::do_put`] and [`FlightClient::do_exchange`] from the client
+/// provided input stream to the response stream.
+///
+/// # Error Priority
+/// Error from the receiver are prioritised over the response stream.
+///
+/// [`FlightClient::do_put`]: crate::FlightClient::do_put
+/// [`FlightClient::do_exchange`]: crate::FlightClient::do_exchange
+pub(crate) struct FallibleTonicResponseStream<T> {
+    /// Receiver for FlightError
+    receiver: Receiver<FlightError>,
+    /// Tonic response stream
+    response_stream: Pin<Box<dyn Stream<Item = Result<T, tonic::Status>> + Send + 'static>>,
+}
+
+impl<T> FallibleTonicResponseStream<T> {
+    pub(crate) fn new(
+        receiver: Receiver<FlightError>,
+        response_stream: Pin<Box<dyn Stream<Item = Result<T, tonic::Status>> + Send + 'static>>,
+    ) -> Self {
+        Self {
+            receiver,
+            response_stream,
+        }
+    }
+}
+
+impl<T> Stream for FallibleTonicResponseStream<T> {
+    type Item = Result<T, FlightError>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        let pinned = self.get_mut();
+        let receiver = &mut pinned.receiver;
+        // Prioritise sending the error that's been notified over
+        // polling the response_stream
+        if let Poll::Ready(Ok(err)) = receiver.poll_unpin(cx) {
+            return Poll::Ready(Some(Err(err)));
+        };
+
+        match ready!(pinned.response_stream.poll_next_unpin(cx)) {
+            Some(Ok(res)) => Poll::Ready(Some(Ok(res))),
+            Some(Err(status)) => Poll::Ready(Some(Err(FlightError::Tonic(status)))),
+            None => Poll::Ready(None),
+        }
+    }
+}
diff --git a/arrow-schema/src/ffi.rs b/arrow-schema/src/ffi.rs
index b0c6a9f52..c68849059 100644
--- a/arrow-schema/src/ffi.rs
+++ b/arrow-schema/src/ffi.rs
@@ -261,12 +261,17 @@ impl FFI_ArrowSchema {
     }
 
     /// returns the name of this schema.
-    pub fn name(&self) -> &str {
-        assert!(!self.name.is_null());
-        // safe because the lifetime of `self.name` equals `self`
-        unsafe { CStr::from_ptr(self.name) }
-            .to_str()
-            .expect("The external API has a non-utf8 as name")
+    pub fn name(&self) -> Option<&str> {
+        if self.name.is_null() {
+            None
+        } else {
+            // safe because the lifetime of `self.name` equals `self`
+            Some(
+                unsafe { CStr::from_ptr(self.name) }
+                    .to_str()
+                    .expect("The external API has a non-utf8 as name"),
+            )
+        }
     }
 
     pub fn flags(&self) -> Option<Flags> {
@@ -582,7 +587,7 @@ impl TryFrom<&FFI_ArrowSchema> for Field {
 
     fn try_from(c_schema: &FFI_ArrowSchema) -> Result<Self, ArrowError> {
         let dtype = DataType::try_from(c_schema)?;
-        let mut field = Field::new(c_schema.name(), dtype, c_schema.nullable());
+        let mut field = Field::new(c_schema.name().unwrap_or(""), dtype, c_schema.nullable());
         field.set_metadata(c_schema.metadata()?);
         Ok(field)
     }
@@ -918,4 +923,13 @@ mod tests {
             assert_eq!(field.metadata(), &metadata);
         }
     }
+
+    #[test]
+    fn test_import_field_with_null_name() {
+        let dtype = DataType::Int16;
+        let c_schema = FFI_ArrowSchema::try_from(&dtype).unwrap();
+        assert!(c_schema.name().is_none());
+        let field = Field::try_from(&c_schema).unwrap();
+        assert_eq!(field.name(), "");
+    }
 }
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 007e27108..6fe4889db 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -705,7 +705,14 @@ impl GetClient for S3Client {
         };
 
         let mut builder = self.client.request(method, url);
-        builder = builder.headers(self.config.encryption_headers.clone().into());
+        if self
+            .config
+            .encryption_headers
+            .0
+            .contains_key("x-amz-server-side-encryption-customer-algorithm")
+        {
+            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/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/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