Repository: apache/arrow-rs
Pull request #6350 (2024-09-02 13:25:35)

Update prost-build requirement from =0.13.1 to =0.13.2
Updates the requirements on [prost-build](https://github.com/tokio-rs/prost) to permit the latest version.
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/tokio-rs/prost/blob/master/CHANGELOG.md">prost-build's changelog</a>.</em></p>
<blockquote>
<h1>Prost version 0.13.2</h1>
<p><em>PROST!</em> is a <a href="https://developers.google.com/protocol-buffers/">Protocol Buffers</a> implementation for the <a href="https://www.rust-lang.org/">Rust Language</a>. <code>prost</code> generates simple, idiomatic Rust code from <code>proto2</code> and <code>proto3</code> files.</p>
<h2>Features</h2>
<ul>
<li>prost-build: Add protoc executable path to Config (<a href="https://redirect.github.com/tokio-rs/prost/issues/1126">#1126</a>)</li>
<li>prost-build: Extract file descriptor loading from compile_protos() (<a href="https://redirect.github.com/tokio-rs/prost/issues/1067">#1067</a>)</li>
</ul>
<h2>Bug Fixes</h2>
<ul>
<li>prost-types: Fix date-time parsing (<a href="https://redirect.github.com/tokio-rs/prost/issues/1096">#1096</a>)</li>
<li>prost-types: '+' is not a numeric digit (<a href="https://redirect.github.com/tokio-rs/prost/issues/1104">#1104</a>)</li>
<li>prost-types: Converting DateTime to Timestamp is fallible (<a href="https://redirect.github.com/tokio-rs/prost/issues/1095">#1095</a>)</li>
<li>prost-types: Parse timestamp with long second fraction (<a href="https://redirect.github.com/tokio-rs/prost/issues/1106">#1106</a>)</li>
<li>prost-types: Format negative fractional duration (<a href="https://redirect.github.com/tokio-rs/prost/issues/1110">#1110</a>)</li>
<li>prost-types: Allow unknown local time offset (<a href="https://redirect.github.com/tokio-rs/prost/issues/1109">#1109</a>)</li>
</ul>
<h2>Styling</h2>
<ul>
<li>Remove use of legacy numeric constants (<a href="https://redirect.github.com/tokio-rs/prost/issues/1089">#1089</a>)</li>
<li>Move encoding functions into separate modules (<a href="https://redirect.github.com/tokio-rs/prost/issues/1111">#1111</a>)</li>
<li>Remove needless borrow (<a href="https://redirect.github.com/tokio-rs/prost/issues/1122">#1122</a>)</li>
</ul>
<h2>Testing</h2>
<ul>
<li>Add tests for public interface of DecodeError (<a href="https://redirect.github.com/tokio-rs/prost/issues/1120">#1120</a>)</li>
<li>Add <code>parse_date</code> fuzzing target (<a href="https://redirect.github.com/tokio-rs/prost/issues/1127">#1127</a>)</li>
<li>Fix build without std (<a href="https://redirect.github.com/tokio-rs/prost/issues/1134">#1134</a>)</li>
<li>Change some proptest to kani proofs (<a href="https://redirect.github.com/tokio-rs/prost/issues/1133">#1133</a>)</li>
<li>Add <code>parse_duration</code> fuzzing target (<a href="https://redirect.github.com/tokio-rs/prost/issues/1129">#1129</a>)</li>
<li>fuzz: Fix building of fuzzing targets (<a href="https://redirect.github.com/tokio-rs/prost/issues/1107">#1107</a>)</li>
<li>fuzz: Add fuzz targets to workspace (<a href="https://redirect.github.com/tokio-rs/prost/issues/1117">#1117</a>)</li>
</ul>
<h2>Miscellaneous Tasks</h2>
<ul>
<li>Move old protobuf benchmark into prost (<a href="https://redirect.github.com/tokio-rs/prost/issues/1100">#1100</a>)</li>
<li>Remove allow clippy::derive_partial_eq_without_eq (<a href="https://redirect.github.com/tokio-rs/prost/issues/1115">#1115</a>)</li>
<li>Run <code>cargo test</code> without <code>all-targets</code> (<a href="https://redirect.github.com/tokio-rs/prost/issues/1118">#1118</a>)</li>
<li>dependabot: Add github actions (<a href="https://redirect.github.com/tokio-rs/prost/issues/1121">#1121</a>)</li>
<li>Update to cargo clippy version 1.80 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1128">#1128</a>)</li>
</ul>
<h2>Build</h2>
<ul>
<li>Use <code>proc-macro</code> in Cargo.toml (<a href="https://redirect.github.com/tokio-rs/prost/issues/1102">#1102</a>)</li>
<li>Ignore missing features in <code>tests</code> crates (<a href="https://redirect.github.com/tokio-rs/prost/issues/1101">#1101</a>)</li>
<li>Use separated build directory for protobuf (<a href="https://redirect.github.com/tokio-rs/prost/issues/1103">#1103</a>)</li>
<li>protobuf: Don't install unused test proto (<a href="https://redirect.github.com/tokio-rs/prost/issues/1116">#1116</a>)</li>
<li>protobuf: Use crate <code>cmake</code> (<a href="https://redirect.github.com/tokio-rs/prost/issues/1137">#1137</a>)</li>
<li>deps: Update devcontainer to Debian Bookworm release (<a href="https://redirect.github.com/tokio-rs/prost/issues/1114">#1114</a>)</li>
</ul>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/tokio-rs/prost/commit/57e794203106db483e5115e7e67502ef6f2c7ad8"><code>57e7942</code></a> chore: Release version 0.13.2 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1139">#1139</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/8424775d78b13239df3cf3fe888236770a0cd839"><code>8424775</code></a> build(protobuf): Use crate <code>cmake</code> (<a href="https://redirect.github.com/tokio-rs/prost/issues/1137">#1137</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/21208abf667313866f79d3d1438310c4dc20bdff"><code>21208ab</code></a> build(deps): bump model-checking/kani-github-action from 0.32 to 1.1 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1125">#1125</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/0c79864443621f20d92f9acc78a6ab0e7821dab0"><code>0c79864</code></a> tests(fuzz): Add <code>parse_duration</code> fuzzing target (<a href="https://redirect.github.com/tokio-rs/prost/issues/1129">#1129</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/52046b943fdf6f79461725027245f890c7b4f514"><code>52046b9</code></a> tests: Change some proptest to kani proofs (<a href="https://redirect.github.com/tokio-rs/prost/issues/1133">#1133</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/ee59dd5a9fe0935ad50e6ddbea5d23e3c6419468"><code>ee59dd5</code></a> tests: Fix build without std (<a href="https://redirect.github.com/tokio-rs/prost/issues/1134">#1134</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/e773f5f6d38f74d0efff876011a2fd0d002aed4c"><code>e773f5f</code></a> feat(prost-build): Add protoc executable path to Config (<a href="https://redirect.github.com/tokio-rs/prost/issues/1126">#1126</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/753bd92a85a3aa305d9d96b5c6363dc58d6356e6"><code>753bd92</code></a> ci(clippy): Update to cargo clippy version 1.80 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1128">#1128</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/df3e58e5d113a0dcf8b6735a5d04cde2d74e5df3"><code>df3e58e</code></a> tests(fuzz): Add <code>parse_date</code> fuzzing target (<a href="https://redirect.github.com/tokio-rs/prost/issues/1127">#1127</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/409b93214ed8d98fbb364031ccf330ce4e7caa32"><code>409b932</code></a> style: Remove needless borrow (<a href="https://redirect.github.com/tokio-rs/prost/issues/1122">#1122</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/tokio-rs/prost/compare/v0.13.1...v0.13.2">compare view</a></li>
</ul>
</details>
<br />


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index a1644ee49..1b65c5057 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -100,6 +100,13 @@ jobs:
         run: rustup component add rustfmt
       - name: Format arrow
         run: cargo fmt --all -- --check
+      - name: Format parquet
+        # Many modules in parquet are skipped, so check parquet separately. If this check fails, run:
+        #   cargo fmt -p parquet -- --config skip_children=true `find ./parquet -name "*.rs" \! -name format.rs`
+        # from the top level arrow-rs directory and check in the result.
+        # https://github.com/apache/arrow-rs/issues/6179
+        working-directory: parquet
+        run: cargo fmt -p parquet -- --check --config skip_children=true `find . -name "*.rs" \! -name format.rs`
       - name: Format object_store
         working-directory: object_store
         run: cargo fmt --all -- --check
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 72b809019..20c13ffeb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -32,6 +32,7 @@
 - Upgrade protobuf definitions to flightsql 17.0 \(\#6133\) [\#6169](https://github.com/apache/arrow-rs/pull/6169) [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([alamb](https://github.com/alamb))
 - Remove automatic buffering in `ipc::reader::FileReader` for for consistent buffering [\#6132](https://github.com/apache/arrow-rs/pull/6132) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([V0ldek](https://github.com/V0ldek))
 - No longer write Parquet column metadata after column chunks \*and\* in the footer [\#6117](https://github.com/apache/arrow-rs/pull/6117) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Remove `impl<T: AsRef<[u8]>> From<T> for Buffer` that easily accidentally copies data [\#6043](https://github.com/apache/arrow-rs/pull/6043) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
 
 **Implemented enhancements:**
 
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index a53604d0f..e0adc18a9 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -160,6 +160,13 @@ PR be sure to run the following and check for lint issues:
 cargo +stable fmt --all -- --check
 ```
 
+Note that currently the above will not check all source files in the parquet crate. To check all
+parquet files run the following from the top-level `arrow-rs` directory:
+
+```bash
+cargo fmt -p parquet -- --check --config skip_children=true `find . -name "*.rs" \! -name format.rs`
+```
+
 ## Breaking Changes
 
 Our [release schedule] allows breaking API changes only in major releases.
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index 6170db85d..57b86c192 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -45,7 +45,7 @@ arrow-buffer = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-data = { workspace = true }
 chrono = { workspace = true }
-chrono-tz = { version = "0.9", optional = true }
+chrono-tz = { version = "0.10", optional = true }
 num = { version = "0.4.1", default-features = false, features = ["std"] }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
 hashbrown = { version = "0.14.2", default-features = false }
diff --git a/arrow-array/src/builder/generic_bytes_builder.rs b/arrow-array/src/builder/generic_bytes_builder.rs
index 516f45be3..e85a2fed0 100644
--- a/arrow-array/src/builder/generic_bytes_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_builder.rs
@@ -60,8 +60,9 @@ impl<T: ByteArrayType> GenericByteBuilder<T> {
     /// Creates a new  [`GenericByteBuilder`] from buffers.
     ///
     /// # Safety
-    /// This doesn't verify buffer contents as it assumes the buffers are from existing and
-    /// valid [`GenericByteArray`].
+    ///
+    /// This doesn't verify buffer contents as it assumes the buffers are from
+    /// existing and valid [`GenericByteArray`].
     pub unsafe fn new_from_buffer(
         offsets_buffer: MutableBuffer,
         value_buffer: MutableBuffer,
@@ -88,9 +89,19 @@ impl<T: ByteArrayType> GenericByteBuilder<T> {
 
     /// Appends a value into the builder.
     ///
+    /// See the [GenericStringBuilder] documentation for examples of
+    /// incrementally building string values with multiple `write!` calls.
+    ///
     /// # Panics
     ///
-    /// Panics if the resulting length of [`Self::values_slice`] would exceed `T::Offset::MAX`
+    /// Panics if the resulting length of [`Self::values_slice`] would exceed
+    /// `T::Offset::MAX` bytes.
+    ///
+    /// For example, this can happen with [`StringArray`] or [`BinaryArray`]
+    /// where the total length of all values exceeds 2GB
+    ///
+    /// [`StringArray`]: crate::StringArray
+    /// [`BinaryArray`]: crate::BinaryArray
     #[inline]
     pub fn append_value(&mut self, value: impl AsRef<T::Native>) {
         self.value_builder.append_slice(value.as_ref().as_ref());
@@ -99,6 +110,11 @@ impl<T: ByteArrayType> GenericByteBuilder<T> {
     }
 
     /// Append an `Option` value into the builder.
+    ///
+    /// - A `None` value will append a null value.
+    /// - A `Some` value will append the value.
+    ///
+    /// See [`Self::append_value`] for more panic information.
     #[inline]
     pub fn append_option(&mut self, value: Option<impl AsRef<T::Native>>) {
         match value {
@@ -227,30 +243,51 @@ impl<T: ByteArrayType, V: AsRef<T::Native>> Extend<Option<V>> for GenericByteBui
 /// Values can be appended using [`GenericByteBuilder::append_value`], and nulls with
 /// [`GenericByteBuilder::append_null`].
 ///
-/// Additionally, implements [`std::fmt::Write`] with any written data included in the next
-/// appended value. This allows use with [`std::fmt::Display`] without intermediate allocations
+/// This builder also implements [`std::fmt::Write`] with any written data
+/// included in the next appended value. This allows using [`std::fmt::Display`]
+/// with standard Rust idioms like `write!` and `writeln!` to write data
+/// directly to the builder without intermediate allocations.
+///
+/// # Example writing strings with `append_value`
+/// ```
+/// # use arrow_array::builder::GenericStringBuilder;
+/// let mut builder = GenericStringBuilder::<i32>::new();
+///
+/// // Write one string value
+/// builder.append_value("foobarbaz");
+///
+/// // Write a second string
+/// builder.append_value("v2");
+///
+/// let array = builder.finish();
+/// assert_eq!(array.value(0), "foobarbaz");
+/// assert_eq!(array.value(1), "v2");
+/// ```
+///
+/// # Example incrementally writing strings with `std::fmt::Write`
 ///
-/// # Example
 /// ```
 /// # use std::fmt::Write;
 /// # use arrow_array::builder::GenericStringBuilder;
 /// let mut builder = GenericStringBuilder::<i32>::new();
 ///
-/// // Write data
+/// // Write data in multiple `write!` calls
 /// write!(builder, "foo").unwrap();
 /// write!(builder, "bar").unwrap();
-///
-/// // Finish value
+/// // The next call to append_value finishes the current string
+/// // including all previously written strings.
 /// builder.append_value("baz");
 ///
-/// // Write second value
+/// // Write second value with a single write call
 /// write!(builder, "v2").unwrap();
+/// // finish the value by calling append_value with an empty string
 /// builder.append_value("");
 ///
 /// let array = builder.finish();
 /// assert_eq!(array.value(0), "foobarbaz");
 /// assert_eq!(array.value(1), "v2");
 /// ```
+///
 pub type GenericStringBuilder<O> = GenericByteBuilder<GenericStringType<O>>;
 
 impl<O: OffsetSizeTrait> Write for GenericStringBuilder<O> {
diff --git a/arrow-buffer/Cargo.toml b/arrow-buffer/Cargo.toml
index 8bc33b187..68bfe8ddf 100644
--- a/arrow-buffer/Cargo.toml
+++ b/arrow-buffer/Cargo.toml
@@ -44,10 +44,14 @@ rand = { version = "0.8", default-features = false, features = ["std", "std_rng"
 
 [build-dependencies]
 
+[[bench]]
+name = "bit_mask"
+harness = false
+
 [[bench]]
 name = "i256"
 harness = false
 
 [[bench]]
 name = "offset"
-harness = false
\ No newline at end of file
+harness = false
diff --git a/arrow-buffer/benches/bit_mask.rs b/arrow-buffer/benches/bit_mask.rs
new file mode 100644
index 000000000..6907e336a
--- /dev/null
+++ b/arrow-buffer/benches/bit_mask.rs
@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow_buffer::bit_mask::set_bits;
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("bit_mask");
+
+    for offset_write in [0, 5] {
+        for offset_read in [0, 5] {
+            for len in [1, 17, 65] {
+                for datum in [0u8, 0xADu8] {
+                    let x = (offset_write, offset_read, len, datum);
+                    group.bench_with_input(
+                        BenchmarkId::new(
+                            "set_bits",
+                            format!(
+                                "offset_write_{}_offset_read_{}_len_{}_datum_{}",
+                                x.0, x.1, x.2, x.3
+                            ),
+                        ),
+                        &x,
+                        |b, &x| {
+                            b.iter(|| {
+                                set_bits(
+                                    black_box(&mut [0u8; 9]),
+                                    black_box(&[x.3; 9]),
+                                    black_box(x.0),
+                                    black_box(x.1),
+                                    black_box(x.2),
+                                )
+                            });
+                        },
+                    );
+                }
+            }
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
diff --git a/arrow-buffer/src/buffer/immutable.rs b/arrow-buffer/src/buffer/immutable.rs
index c53ef18ba..7cd355221 100644
--- a/arrow-buffer/src/buffer/immutable.rs
+++ b/arrow-buffer/src/buffer/immutable.rs
@@ -382,6 +382,12 @@ impl<const N: usize> From<&[u8; N]> for Buffer {
     }
 }
 
+impl<T: ArrowNativeType> From<Vec<T>> for Buffer {
+    fn from(value: Vec<T>) -> Self {
+        Self::from_vec(value)
+    }
+}
+
 /// Creating a `Buffer` instance by storing the boolean values into the buffer
 impl FromIterator<bool> for Buffer {
     fn from_iter<I>(iter: I) -> Self
diff --git a/arrow-data/src/ffi.rs b/arrow-data/src/ffi.rs
index 589f7dac6..3345595fa 100644
--- a/arrow-data/src/ffi.rs
+++ b/arrow-data/src/ffi.rs
@@ -324,6 +324,6 @@ mod tests {
 
         assert_eq!(0, private_data.buffers_ptr.len());
 
-        Box::into_raw(private_data);
+        let _ = Box::into_raw(private_data);
     }
 }
diff --git a/arrow-flight/gen/Cargo.toml b/arrow-flight/gen/Cargo.toml
index ae6000ae7..e4801cdaa 100644
--- a/arrow-flight/gen/Cargo.toml
+++ b/arrow-flight/gen/Cargo.toml
@@ -33,5 +33,5 @@ publish = false
 # Pin specific version of the tonic-build dependencies to avoid auto-generated
 # (and checked in) arrow.flight.protocol.rs from changing
 proc-macro2 = { version = "=1.0.86", default-features = false }
-prost-build = { version = "=0.13.1", default-features = false }
+prost-build = { version = "=0.13.2", default-features = false }
 tonic-build = { version = "=0.12.2", default-features = false, features = ["transport", "prost"] }
diff --git a/arrow-flight/src/arrow.flight.protocol.rs b/arrow-flight/src/arrow.flight.protocol.rs
index 670d4a649..f3ae7774a 100644
--- a/arrow-flight/src/arrow.flight.protocol.rs
+++ b/arrow-flight/src/arrow.flight.protocol.rs
@@ -3,7 +3,6 @@
 // This file is @generated by prost-build.
 ///
 /// The request that a client provides to a server on handshake.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct HandshakeRequest {
     ///
@@ -15,7 +14,6 @@ pub struct HandshakeRequest {
     #[prost(bytes = "bytes", tag = "2")]
     pub payload: ::prost::bytes::Bytes,
 }
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct HandshakeResponse {
     ///
@@ -29,7 +27,6 @@ pub struct HandshakeResponse {
 }
 ///
 /// A message for doing simple auth.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BasicAuth {
     #[prost(string, tag = "2")]
@@ -37,13 +34,11 @@ pub struct BasicAuth {
     #[prost(string, tag = "3")]
     pub password: ::prost::alloc::string::String,
 }
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct Empty {}
 ///
 /// Describes an available action, including both the name used for execution
 /// along with a short description of the purpose of the action.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionType {
     #[prost(string, tag = "1")]
@@ -54,7 +49,6 @@ pub struct ActionType {
 ///
 /// A service specific expression that can be used to return a limited set
 /// of available Arrow Flight streams.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Criteria {
     #[prost(bytes = "bytes", tag = "1")]
@@ -62,7 +56,6 @@ pub struct Criteria {
 }
 ///
 /// An opaque action specific for the service.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Action {
     #[prost(string, tag = "1")]
@@ -74,7 +67,6 @@ pub struct Action {
 /// The request of the CancelFlightInfo action.
 ///
 /// The request should be stored in Action.body.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CancelFlightInfoRequest {
     #[prost(message, optional, tag = "1")]
@@ -84,7 +76,6 @@ pub struct CancelFlightInfoRequest {
 /// The request of the RenewFlightEndpoint action.
 ///
 /// The request should be stored in Action.body.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct RenewFlightEndpointRequest {
     #[prost(message, optional, tag = "1")]
@@ -92,7 +83,6 @@ pub struct RenewFlightEndpointRequest {
 }
 ///
 /// An opaque result returned after executing an action.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Result {
     #[prost(bytes = "bytes", tag = "1")]
@@ -102,7 +92,6 @@ pub struct Result {
 /// The result of the CancelFlightInfo action.
 ///
 /// The result should be stored in Result.body.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CancelFlightInfoResult {
     #[prost(enumeration = "CancelStatus", tag = "1")]
@@ -110,7 +99,6 @@ pub struct CancelFlightInfoResult {
 }
 ///
 /// Wrap the result of a getSchema call
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SchemaResult {
     /// The schema of the dataset in its IPC form:
@@ -123,7 +111,6 @@ pub struct SchemaResult {
 ///
 /// The name or tag for a Flight. May be used as a way to retrieve or generate
 /// a flight or be used to expose a set of previously defined flights.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightDescriptor {
     #[prost(enumeration = "flight_descriptor::DescriptorType", tag = "1")]
@@ -193,7 +180,6 @@ pub mod flight_descriptor {
 ///
 /// The access coordinates for retrieval of a dataset. With a FlightInfo, a
 /// consumer is able to determine how to retrieve a dataset.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightInfo {
     /// The schema of the dataset in its IPC form:
@@ -253,7 +239,6 @@ pub struct FlightInfo {
 }
 ///
 /// The information to process a long-running query.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PollInfo {
     ///
@@ -294,7 +279,6 @@ pub struct PollInfo {
 }
 ///
 /// A particular stream or split associated with a flight.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightEndpoint {
     ///
@@ -338,7 +322,6 @@ pub struct FlightEndpoint {
 ///
 /// A location where a Flight service will accept retrieval of a particular
 /// stream given a ticket.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Location {
     #[prost(string, tag = "1")]
@@ -350,7 +333,6 @@ pub struct Location {
 ///
 /// Tickets are meant to be single use. It is an error/application-defined
 /// behavior to reuse a ticket.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Ticket {
     #[prost(bytes = "bytes", tag = "1")]
@@ -358,7 +340,6 @@ pub struct Ticket {
 }
 ///
 /// A batch of Arrow data as part of a stream of batches.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightData {
     ///
@@ -384,7 +365,6 @@ pub struct FlightData {
 }
 /// *
 /// The response message associated with the submission of a DoPut.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PutResult {
     #[prost(bytes = "bytes", tag = "1")]
diff --git a/arrow-flight/src/bin/flight_sql_client.rs b/arrow-flight/src/bin/flight_sql_client.rs
index 296efc1c3..c334b95a9 100644
--- a/arrow-flight/src/bin/flight_sql_client.rs
+++ b/arrow-flight/src/bin/flight_sql_client.rs
@@ -20,7 +20,10 @@ use std::{sync::Arc, time::Duration};
 use anyhow::{bail, Context, Result};
 use arrow_array::{ArrayRef, Datum, RecordBatch, StringArray};
 use arrow_cast::{cast_with_options, pretty::pretty_format_batches, CastOptions};
-use arrow_flight::{sql::client::FlightSqlServiceClient, FlightInfo};
+use arrow_flight::{
+    sql::{client::FlightSqlServiceClient, CommandGetDbSchemas, CommandGetTables},
+    FlightInfo,
+};
 use arrow_schema::Schema;
 use clap::{Parser, Subcommand};
 use futures::TryStreamExt;
@@ -111,6 +114,51 @@ struct Args {
 /// Different available commands.
 #[derive(Debug, Subcommand)]
 enum Command {
+    /// Get catalogs.
+    Catalogs,
+    /// Get db schemas for a catalog.
+    DbSchemas {
+        /// Name of a catalog.
+        ///
+        /// Required.
+        catalog: String,
+        /// Specifies a filter pattern for schemas to search for.
+        /// When no schema_filter is provided, the pattern will not be used to narrow the search.
+        /// In the pattern string, two special characters can be used to denote matching rules:
+        ///     - "%" means to match any substring with 0 or more characters.
+        ///     - "_" means to match any one character.
+        #[clap(short, long)]
+        db_schema_filter: Option<String>,
+    },
+    /// Get tables for a catalog.
+    Tables {
+        /// Name of a catalog.
+        ///
+        /// Required.
+        catalog: String,
+        /// Specifies a filter pattern for schemas to search for.
+        /// When no schema_filter is provided, the pattern will not be used to narrow the search.
+        /// In the pattern string, two special characters can be used to denote matching rules:
+        ///     - "%" means to match any substring with 0 or more characters.
+        ///     - "_" means to match any one character.
+        #[clap(short, long)]
+        db_schema_filter: Option<String>,
+        /// Specifies a filter pattern for tables to search for.
+        /// When no table_filter is provided, all tables matching other filters are searched.
+        /// In the pattern string, two special characters can be used to denote matching rules:
+        ///     - "%" means to match any substring with 0 or more characters.
+        ///     - "_" means to match any one character.
+        #[clap(short, long)]
+        table_filter: Option<String>,
+        /// Specifies a filter of table types which must match.
+        /// The table types depend on vendor/implementation. It is usually used to separate tables from views or system tables.
+        /// TABLE, VIEW, and SYSTEM TABLE are commonly supported.
+        #[clap(long)]
+        table_types: Vec<String>,
+    },
+    /// Get table types.
+    TableTypes,
+
     /// Execute given statement.
     StatementQuery {
         /// SQL query.
@@ -150,6 +198,36 @@ async fn main() -> Result<()> {
         .context("setup client")?;
 
     let flight_info = match args.cmd {
+        Command::Catalogs => client.get_catalogs().await.context("get catalogs")?,
+        Command::DbSchemas {
+            catalog,
+            db_schema_filter,
+        } => client
+            .get_db_schemas(CommandGetDbSchemas {
+                catalog: Some(catalog),
+                db_schema_filter_pattern: db_schema_filter,
+            })
+            .await
+            .context("get db schemas")?,
+        Command::Tables {
+            catalog,
+            db_schema_filter,
+            table_filter,
+            table_types,
+        } => client
+            .get_tables(CommandGetTables {
+                catalog: Some(catalog),
+                db_schema_filter_pattern: db_schema_filter,
+                table_name_filter_pattern: table_filter,
+                table_types,
+                // Schema is returned as ipc encoded bytes.
+                // We do not support returning the schema as there is no trivial mechanism
+                // to display the information to the user.
+                include_schema: false,
+            })
+            .await
+            .context("get tables")?,
+        Command::TableTypes => client.get_table_types().await.context("get table types")?,
         Command::StatementQuery { query } => client
             .execute(query, None)
             .await
diff --git a/arrow-flight/src/sql/arrow.flight.protocol.sql.rs b/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
index 3eeed6ff4..805759cde 100644
--- a/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
+++ b/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
@@ -19,7 +19,6 @@
 ///               int32_to_int32_list_map: map<key: int32, value: list<$data$: int32>>
 /// >
 /// where there is one row per requested piece of metadata information.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetSqlInfo {
     ///
@@ -100,7 +99,6 @@ pub struct CommandGetSqlInfo {
 ///                               is only relevant to be used by ODBC).
 /// >
 /// The returned data should be ordered by data_type and then by type_name.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CommandGetXdbcTypeInfo {
     ///
@@ -120,7 +118,6 @@ pub struct CommandGetXdbcTypeInfo {
 ///   catalog_name: utf8 not null
 /// >
 /// The returned data should be ordered by catalog_name.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CommandGetCatalogs {}
 ///
@@ -136,7 +133,6 @@ pub struct CommandGetCatalogs {}
 ///   db_schema_name: utf8 not null
 /// >
 /// The returned data should be ordered by catalog_name, then db_schema_name.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetDbSchemas {
     ///
@@ -181,7 +177,6 @@ pub struct CommandGetDbSchemas {
 ///   - ARROW:FLIGHT:SQL:IS_READ_ONLY      - "1" indicates if the column is read only, "0" otherwise.
 ///   - ARROW:FLIGHT:SQL:IS_SEARCHABLE     - "1" indicates if the column is searchable via WHERE clause, "0" otherwise.
 /// The returned data should be ordered by catalog_name, db_schema_name, table_name, then table_type, followed by table_schema if requested.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetTables {
     ///
@@ -231,7 +226,6 @@ pub struct CommandGetTables {
 ///   table_type: utf8 not null
 /// >
 /// The returned data should be ordered by table_type.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CommandGetTableTypes {}
 ///
@@ -250,7 +244,6 @@ pub struct CommandGetTableTypes {}
 ///   key_sequence: int32 not null
 /// >
 /// The returned data should be ordered by catalog_name, db_schema_name, table_name, key_name, then key_sequence.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetPrimaryKeys {
     ///
@@ -294,7 +287,6 @@ pub struct CommandGetPrimaryKeys {
 /// >
 /// The returned data should be ordered by fk_catalog_name, fk_db_schema_name, fk_table_name, fk_key_name, then key_sequence.
 /// update_rule and delete_rule returns a byte that is equivalent to actions declared on UpdateDeleteRules enum.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetExportedKeys {
     ///
@@ -342,7 +334,6 @@ pub struct CommandGetExportedKeys {
 ///     - 2 = SET NULL
 ///     - 3 = NO ACTION
 ///     - 4 = SET DEFAULT
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetImportedKeys {
     ///
@@ -392,7 +383,6 @@ pub struct CommandGetImportedKeys {
 ///     - 2 = SET NULL
 ///     - 3 = NO ACTION
 ///     - 4 = SET DEFAULT
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetCrossReference {
     /// *
@@ -430,7 +420,6 @@ pub struct CommandGetCrossReference {
 }
 ///
 /// Request message for the "CreatePreparedStatement" action on a Flight SQL enabled backend.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCreatePreparedStatementRequest {
     /// The valid SQL string to create a prepared statement for.
@@ -443,7 +432,6 @@ pub struct ActionCreatePreparedStatementRequest {
 }
 ///
 /// An embedded message describing a Substrait plan to execute.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SubstraitPlan {
     /// The serialized substrait.Plan to create a prepared statement for.
@@ -460,7 +448,6 @@ pub struct SubstraitPlan {
 }
 ///
 /// Request message for the "CreatePreparedSubstraitPlan" action on a Flight SQL enabled backend.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCreatePreparedSubstraitPlanRequest {
     /// The serialized substrait.Plan to create a prepared statement for.
@@ -479,7 +466,6 @@ pub struct ActionCreatePreparedSubstraitPlanRequest {
 /// - Automatically, by a server timeout.
 ///
 /// The result should be wrapped in a google.protobuf.Any message.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCreatePreparedStatementResult {
     /// Opaque handle for the prepared statement on the server.
@@ -500,7 +486,6 @@ pub struct ActionCreatePreparedStatementResult {
 ///
 /// Request message for the "ClosePreparedStatement" action on a Flight SQL enabled backend.
 /// Closes server resources associated with the prepared statement handle.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionClosePreparedStatementRequest {
     /// Opaque handle for the prepared statement on the server.
@@ -510,7 +495,6 @@ pub struct ActionClosePreparedStatementRequest {
 ///
 /// Request message for the "BeginTransaction" action.
 /// Begins a transaction.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ActionBeginTransactionRequest {}
 ///
@@ -519,7 +503,6 @@ pub struct ActionBeginTransactionRequest {}
 ///
 /// Only supported if FLIGHT_SQL_TRANSACTION is
 /// FLIGHT_SQL_TRANSACTION_SUPPORT_SAVEPOINT.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionBeginSavepointRequest {
     /// The transaction to which a savepoint belongs.
@@ -537,7 +520,6 @@ pub struct ActionBeginSavepointRequest {
 /// automatically rolled back.
 ///
 /// The result should be wrapped in a google.protobuf.Any message.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionBeginTransactionResult {
     /// Opaque handle for the transaction on the server.
@@ -552,7 +534,6 @@ pub struct ActionBeginTransactionResult {
 /// out, then the savepoint is also invalidated.
 ///
 /// The result should be wrapped in a google.protobuf.Any message.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionBeginSavepointResult {
     /// Opaque handle for the savepoint on the server.
@@ -566,7 +547,6 @@ pub struct ActionBeginSavepointResult {
 ///
 /// If the action completes successfully, the transaction handle is
 /// invalidated, as are all associated savepoints.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionEndTransactionRequest {
     /// Opaque handle for the transaction on the server.
@@ -629,7 +609,6 @@ pub mod action_end_transaction_request {
 /// Releasing a savepoint invalidates that savepoint.  Rolling back to
 /// a savepoint does not invalidate the savepoint, but invalidates all
 /// savepoints created after the current savepoint.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionEndSavepointRequest {
     /// Opaque handle for the savepoint on the server.
@@ -699,7 +678,6 @@ pub mod action_end_savepoint_request {
 ///     - ARROW:FLIGHT:SQL:IS_READ_ONLY      - "1" indicates if the column is read only, "0" otherwise.
 ///     - ARROW:FLIGHT:SQL:IS_SEARCHABLE     - "1" indicates if the column is searchable via WHERE clause, "0" otherwise.
 ///   - GetFlightInfo: execute the query.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementQuery {
     /// The SQL syntax.
@@ -726,7 +704,6 @@ pub struct CommandStatementQuery {
 ///     - ARROW:FLIGHT:SQL:IS_SEARCHABLE     - "1" indicates if the column is searchable via WHERE clause, "0" otherwise.
 ///   - GetFlightInfo: execute the query.
 ///   - DoPut: execute the query.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementSubstraitPlan {
     /// A serialized substrait.Plan
@@ -739,7 +716,6 @@ pub struct CommandStatementSubstraitPlan {
 /// *
 /// Represents a ticket resulting from GetFlightInfo with a CommandStatementQuery.
 /// This should be used only once and treated as an opaque value, that is, clients should not attempt to parse this.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct TicketStatementQuery {
     /// Unique identifier for the instance of the statement to execute.
@@ -766,7 +742,6 @@ pub struct TicketStatementQuery {
 ///     for the parameters when determining the schema.
 ///   - DoPut: bind parameter values. All of the bound parameter sets will be executed as a single atomic execution.
 ///   - GetFlightInfo: execute the prepared statement instance.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandPreparedStatementQuery {
     /// Opaque handle for the prepared statement on the server.
@@ -776,7 +751,6 @@ pub struct CommandPreparedStatementQuery {
 ///
 /// Represents a SQL update query. Used in the command member of FlightDescriptor
 /// for the RPC call DoPut to cause the server to execute the included SQL update.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementUpdate {
     /// The SQL syntax.
@@ -790,7 +764,6 @@ pub struct CommandStatementUpdate {
 /// Represents a SQL update query. Used in the command member of FlightDescriptor
 /// for the RPC call DoPut to cause the server to execute the included
 /// prepared statement handle as an update.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandPreparedStatementUpdate {
     /// Opaque handle for the prepared statement on the server.
@@ -801,7 +774,6 @@ pub struct CommandPreparedStatementUpdate {
 /// Represents a bulk ingestion request. Used in the command member of FlightDescriptor
 /// for the the RPC call DoPut to cause the server load the contents of the stream's
 /// FlightData into the target destination.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementIngest {
     /// The behavior for handling the table definition.
@@ -838,7 +810,6 @@ pub struct CommandStatementIngest {
 /// Nested message and enum types in `CommandStatementIngest`.
 pub mod command_statement_ingest {
     /// Options for table definition behavior
-    #[allow(clippy::derive_partial_eq_without_eq)]
     #[derive(Clone, Copy, PartialEq, ::prost::Message)]
     pub struct TableDefinitionOptions {
         #[prost(
@@ -949,7 +920,6 @@ pub mod command_statement_ingest {
 /// Returned from the RPC call DoPut when a CommandStatementUpdate,
 /// CommandPreparedStatementUpdate, or CommandStatementIngest was
 /// in the request, containing results from the update.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct DoPutUpdateResult {
     /// The number of records updated. A return value of -1 represents
@@ -962,7 +932,6 @@ pub struct DoPutUpdateResult {
 /// *Note on legacy behavior*: previous versions of the protocol did not return any result for
 /// this command, and that behavior should still be supported by clients. In that case, the client
 /// can continue as though the fields in this message were not provided or set to sensible default values.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct DoPutPreparedStatementResult {
     /// Represents a (potentially updated) opaque handle for the prepared statement on the server.
@@ -992,7 +961,6 @@ pub struct DoPutPreparedStatementResult {
 ///
 /// This command is deprecated since 13.0.0. Use the "CancelFlightInfo"
 /// action with DoAction instead.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCancelQueryRequest {
     /// The result of the GetFlightInfo RPC that initiated the query.
@@ -1009,7 +977,6 @@ pub struct ActionCancelQueryRequest {
 ///
 /// This command is deprecated since 13.0.0. Use the "CancelFlightInfo"
 /// action with DoAction instead.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ActionCancelQueryResult {
     #[prost(enumeration = "action_cancel_query_result::CancelResult", tag = "1")]
diff --git a/arrow-flight/src/sql/metadata/mod.rs b/arrow-flight/src/sql/metadata/mod.rs
index 1e9881ffa..fd71149a3 100644
--- a/arrow-flight/src/sql/metadata/mod.rs
+++ b/arrow-flight/src/sql/metadata/mod.rs
@@ -33,6 +33,7 @@
 mod catalogs;
 mod db_schemas;
 mod sql_info;
+mod table_types;
 mod tables;
 mod xdbc_info;
 
diff --git a/arrow-flight/src/sql/metadata/table_types.rs b/arrow-flight/src/sql/metadata/table_types.rs
new file mode 100644
index 000000000..54cfe6fe2
--- /dev/null
+++ b/arrow-flight/src/sql/metadata/table_types.rs
@@ -0,0 +1,158 @@
+// 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.
+
+//! [`GetTableTypesBuilder`] for building responses to [`CommandGetTableTypes`] queries.
+//!
+//! [`CommandGetTableTypes`]: crate::sql::CommandGetTableTypes
+
+use std::sync::Arc;
+
+use arrow_array::{builder::StringBuilder, ArrayRef, RecordBatch};
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use arrow_select::take::take;
+use once_cell::sync::Lazy;
+
+use crate::error::*;
+use crate::sql::CommandGetTableTypes;
+
+use super::lexsort_to_indices;
+
+/// A builder for a [`CommandGetTableTypes`] response.
+///
+/// Builds rows like this:
+///
+/// * table_type: utf8,
+#[derive(Default)]
+pub struct GetTableTypesBuilder {
+    // array builder for table types
+    table_type: StringBuilder,
+}
+
+impl CommandGetTableTypes {
+    /// Create a builder suitable for constructing a response
+    pub fn into_builder(self) -> GetTableTypesBuilder {
+        self.into()
+    }
+}
+
+impl From<CommandGetTableTypes> for GetTableTypesBuilder {
+    fn from(_value: CommandGetTableTypes) -> Self {
+        Self::new()
+    }
+}
+
+impl GetTableTypesBuilder {
+    /// Create a new instance of [`GetTableTypesBuilder`]
+    pub fn new() -> Self {
+        Self {
+            table_type: StringBuilder::new(),
+        }
+    }
+
+    /// Append a row
+    pub fn append(&mut self, table_type: impl AsRef<str>) {
+        self.table_type.append_value(table_type);
+    }
+
+    /// builds a `RecordBatch` with the correct schema for a `CommandGetTableTypes` response
+    pub fn build(self) -> Result<RecordBatch> {
+        let schema = self.schema();
+        let Self { mut table_type } = self;
+
+        // Make the arrays
+        let table_type = table_type.finish();
+
+        let batch = RecordBatch::try_new(schema, vec![Arc::new(table_type) as ArrayRef])?;
+
+        // Order filtered results by table_type
+        let indices = lexsort_to_indices(batch.columns());
+        let columns = batch
+            .columns()
+            .iter()
+            .map(|c| take(c, &indices, None))
+            .collect::<std::result::Result<Vec<_>, _>>()?;
+
+        Ok(RecordBatch::try_new(batch.schema(), columns)?)
+    }
+
+    /// Return the schema of the RecordBatch that will be returned
+    /// from [`CommandGetTableTypes`]
+    pub fn schema(&self) -> SchemaRef {
+        get_table_types_schema()
+    }
+}
+
+fn get_table_types_schema() -> SchemaRef {
+    Arc::clone(&GET_TABLE_TYPES_SCHEMA)
+}
+
+/// The schema for [`CommandGetTableTypes`].
+static GET_TABLE_TYPES_SCHEMA: Lazy<SchemaRef> = Lazy::new(|| {
+    Arc::new(Schema::new(vec![Field::new(
+        "table_type",
+        DataType::Utf8,
+        false,
+    )]))
+});
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow_array::StringArray;
+
+    fn get_ref_batch() -> RecordBatch {
+        RecordBatch::try_new(
+            get_table_types_schema(),
+            vec![Arc::new(StringArray::from(vec![
+                "a_table_type",
+                "b_table_type",
+                "c_table_type",
+                "d_table_type",
+            ])) as ArrayRef],
+        )
+        .unwrap()
+    }
+
+    #[test]
+    fn test_table_types_are_sorted() {
+        let ref_batch = get_ref_batch();
+
+        let mut builder = GetTableTypesBuilder::new();
+        builder.append("b_table_type");
+        builder.append("a_table_type");
+        builder.append("d_table_type");
+        builder.append("c_table_type");
+        let schema_batch = builder.build().unwrap();
+
+        assert_eq!(schema_batch, ref_batch)
+    }
+
+    #[test]
+    fn test_builder_from_query() {
+        let ref_batch = get_ref_batch();
+        let query = CommandGetTableTypes {};
+
+        let mut builder = query.into_builder();
+        builder.append("a_table_type");
+        builder.append("b_table_type");
+        builder.append("c_table_type");
+        builder.append("d_table_type");
+        let schema_batch = builder.build().unwrap();
+
+        assert_eq!(schema_batch, ref_batch)
+    }
+}
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index ade902f7c..b09dcdc50 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -2572,7 +2572,7 @@ mod tests {
             let mut fields = Vec::new();
             let mut arrays = Vec::new();
             for i in 0..num_cols {
-                let field = Field::new(&format!("col_{}", i), DataType::Decimal128(38, 10), true);
+                let field = Field::new(format!("col_{}", i), DataType::Decimal128(38, 10), true);
                 let array = Decimal128Array::from(vec![num_cols as i128; num_rows]);
                 fields.push(field);
                 arrays.push(Arc::new(array) as Arc<dyn Array>);
@@ -2627,7 +2627,7 @@ mod tests {
         let mut fields = Vec::new();
         let mut arrays = Vec::new();
         for i in 0..num_cols {
-            let field = Field::new(&format!("col_{}", i), DataType::Decimal128(38, 10), true);
+            let field = Field::new(format!("col_{}", i), DataType::Decimal128(38, 10), true);
             let array = Decimal128Array::from(vec![num_cols as i128; num_rows]);
             fields.push(field);
             arrays.push(Arc::new(array) as Arc<dyn Array>);
@@ -2682,7 +2682,7 @@ mod tests {
         let mut fields = Vec::new();
         let options = IpcWriteOptions::try_new(8, false, MetadataVersion::V5).unwrap();
         for i in 0..num_cols {
-            let field = Field::new(&format!("col_{}", i), DataType::Decimal128(38, 10), true);
+            let field = Field::new(format!("col_{}", i), DataType::Decimal128(38, 10), true);
             fields.push(field);
         }
         let schema = Schema::new(fields);
diff --git a/arrow-ord/src/sort.rs b/arrow-ord/src/sort.rs
index 140d878f3..168f82747 100644
--- a/arrow-ord/src/sort.rs
+++ b/arrow-ord/src/sort.rs
@@ -403,7 +403,7 @@ fn sort_fixed_size_list(
 }
 
 #[inline(never)]
-fn sort_impl<T: ?Sized + Copy>(
+fn sort_impl<T: Copy>(
     options: SortOptions,
     valids: &mut [(u32, T)],
     nulls: &[u32],
diff --git a/arrow-schema/src/fields.rs b/arrow-schema/src/fields.rs
index 63aef18dd..5b9ce2a6d 100644
--- a/arrow-schema/src/fields.rs
+++ b/arrow-schema/src/fields.rs
@@ -389,14 +389,13 @@ impl UnionFields {
         let mut set = 0_u128;
         type_ids
             .into_iter()
-            .map(|idx| {
+            .inspect(|&idx| {
                 let mask = 1_u128 << idx;
                 if (set & mask) != 0 {
                     panic!("duplicate type id: {}", idx);
                 } else {
                     set |= mask;
                 }
-                idx
             })
             .zip(fields)
             .collect()
diff --git a/arrow-string/src/length.rs b/arrow-string/src/length.rs
index 79fa46026..82fb2e0d1 100644
--- a/arrow-string/src/length.rs
+++ b/arrow-string/src/length.rs
@@ -48,7 +48,7 @@ fn bit_length_impl<P: ArrowPrimitiveType>(
 /// For list array, length is the number of elements in each list.
 /// For string array and binary array, length is the number of bytes of each value.
 ///
-/// * this only accepts ListArray/LargeListArray, StringArray/LargeStringArray, BinaryArray/LargeBinaryArray, and FixedSizeListArray,
+/// * this only accepts ListArray/LargeListArray, StringArray/LargeStringArray/StringViewArray, BinaryArray/LargeBinaryArray, and FixedSizeListArray,
 ///   or DictionaryArray with above Arrays as values
 /// * length of null is null.
 pub fn length(array: &dyn Array) -> Result<ArrayRef, ArrowError> {
@@ -74,6 +74,14 @@ pub fn length(array: &dyn Array) -> Result<ArrayRef, ArrowError> {
             let list = array.as_string::<i64>();
             Ok(length_impl::<Int64Type>(list.offsets(), list.nulls()))
         }
+        DataType::Utf8View => {
+            let list = array.as_string_view();
+            let v = list.views().iter().map(|v| *v as i32).collect::<Vec<_>>();
+            Ok(Arc::new(PrimitiveArray::<Int32Type>::new(
+                v.into(),
+                list.nulls().cloned(),
+            )))
+        }
         DataType::Binary => {
             let list = array.as_binary::<i32>();
             Ok(length_impl::<Int32Type>(list.offsets(), list.nulls()))
@@ -85,6 +93,14 @@ pub fn length(array: &dyn Array) -> Result<ArrayRef, ArrowError> {
         DataType::FixedSizeBinary(len) | DataType::FixedSizeList(_, len) => Ok(Arc::new(
             Int32Array::new(vec![*len; array.len()].into(), array.nulls().cloned()),
         )),
+        DataType::BinaryView => {
+            let list = array.as_binary_view();
+            let v = list.views().iter().map(|v| *v as i32).collect::<Vec<_>>();
+            Ok(Arc::new(PrimitiveArray::<Int32Type>::new(
+                v.into(),
+                list.nulls().cloned(),
+            )))
+        }
         other => Err(ArrowError::ComputeError(format!(
             "length not supported for {other:?}"
         ))),
@@ -147,9 +163,15 @@ mod tests {
 
     fn length_cases_string() -> Vec<(Vec<&'static str>, usize, Vec<i32>)> {
         // a large array
-        let values = ["one", "on", "o", ""];
+        let values = [
+            "one",
+            "on",
+            "o",
+            "",
+            "this is a longer string to test string array with",
+        ];
         let values = values.into_iter().cycle().take(4096).collect();
-        let expected = [3, 2, 1, 0].into_iter().cycle().take(4096).collect();
+        let expected = [3, 2, 1, 0, 49].into_iter().cycle().take(4096).collect();
 
         vec![
             (vec!["hello", " ", "world"], 3, vec![5, 1, 5]),
@@ -210,6 +232,21 @@ mod tests {
             })
     }
 
+    #[test]
+    fn length_test_string_view() {
+        length_cases_string()
+            .into_iter()
+            .for_each(|(input, len, expected)| {
+                let array = StringViewArray::from(input);
+                let result = length(&array).unwrap();
+                assert_eq!(len, result.len());
+                let result = result.as_any().downcast_ref::<Int32Array>().unwrap();
+                expected.iter().enumerate().for_each(|(i, value)| {
+                    assert_eq!(*value, result.value(i));
+                });
+            })
+    }
+
     #[test]
     fn length_test_binary() {
         let value: Vec<&[u8]> = vec![b"zero", b"one", &[0xff, 0xf8]];
@@ -224,6 +261,23 @@ mod tests {
         length_binary_helper!(i64, Int64Array, length, value, result)
     }
 
+    #[test]
+    fn length_test_binary_view() {
+        let value: Vec<&[u8]> = vec![
+            b"zero",
+            &[0xff, 0xf8],
+            b"two",
+            b"this is a longer string to test binary array with",
+        ];
+        let expected: Vec<i32> = vec![4, 2, 3, 49];
+
+        let array = BinaryViewArray::from(value);
+        let result = length(&array).unwrap();
+        let result = result.as_any().downcast_ref::<Int32Array>().unwrap();
+        let expected: Int32Array = expected.into();
+        assert_eq!(&expected, result);
+    }
+
     #[test]
     fn length_test_list() {
         let value = vec![
diff --git a/conbench/.flake8 b/conbench/.flake8
deleted file mode 100644
index e44b81084..000000000
--- a/conbench/.flake8
+++ /dev/null
@@ -1,2 +0,0 @@
-[flake8]
-ignore = E501
diff --git a/conbench/.gitignore b/conbench/.gitignore
deleted file mode 100755
index aa44ee2ad..000000000
--- a/conbench/.gitignore
+++ /dev/null
@@ -1,130 +0,0 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class
-
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-pip-wheel-metadata/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py,cover
-.hypothesis/
-.pytest_cache/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
-
-# Flask stuff:
-instance/
-.webassets-cache
-
-# Scrapy stuff:
-.scrapy
-
-# Sphinx documentation
-docs/_build/
-
-# PyBuilder
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-.python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-#Pipfile.lock
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
-
diff --git a/conbench/.isort.cfg b/conbench/.isort.cfg
deleted file mode 100644
index f238bf7ea..000000000
--- a/conbench/.isort.cfg
+++ /dev/null
@@ -1,2 +0,0 @@
-[settings]
-profile = black
diff --git a/conbench/README.md b/conbench/README.md
deleted file mode 100644
index 8c7f38cb3..000000000
--- a/conbench/README.md
+++ /dev/null
@@ -1,251 +0,0 @@
-<!---
-  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.
--->
-
-# Arrow Rust + Conbench Integration
-
-
-## Quick start
-
-```
-$ cd ~/arrow-rs/conbench/
-$ conda create -y -n conbench python=3.9
-$ conda activate conbench
-(conbench) $ pip install -r requirements.txt
-(conbench) $ conbench arrow-rs
-```
-
-## Example output
-
-```
-{
-    "batch_id": "b68c559358cc43a3aab02d893d2693f4",
-    "context": {
-        "benchmark_language": "Rust"
-    },
-    "github": {
-        "commit": "ca33a0a50494f95840ade2e9509c3c3d4df35249",
-        "repository": "https://github.com/dianaclarke/arrow-rs"
-    },
-    "info": {},
-    "machine_info": {
-        "architecture_name": "x86_64",
-        "cpu_core_count": "8",
-        "cpu_frequency_max_hz": "2400000000",
-        "cpu_l1d_cache_bytes": "65536",
-        "cpu_l1i_cache_bytes": "131072",
-        "cpu_l2_cache_bytes": "4194304",
-        "cpu_l3_cache_bytes": "0",
-        "cpu_model_name": "Apple M1",
-        "cpu_thread_count": "8",
-        "gpu_count": "0",
-        "gpu_product_names": [],
-        "kernel_name": "20.6.0",
-        "memory_bytes": "17179869184",
-        "name": "diana",
-        "os_name": "macOS",
-        "os_version": "10.16"
-    },
-    "run_id": "08353595bde147fb9deebdb4facd019a",
-    "stats": {
-        "data": [
-            "0.000287",
-            "0.000286",
-            "0.000285",
-            "0.000281",
-            "0.000282",
-            "0.000277",
-            "0.000277",
-            "0.000286",
-            "0.000279",
-            "0.000282",
-            "0.000277",
-            "0.000277",
-            "0.000282",
-            "0.000276",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000284",
-            "0.000288",
-            "0.000278",
-            "0.000276",
-            "0.000278",
-            "0.000275",
-            "0.000275",
-            "0.000275",
-            "0.000275",
-            "0.000281",
-            "0.000284",
-            "0.000277",
-            "0.000277",
-            "0.000278",
-            "0.000282",
-            "0.000281",
-            "0.000284",
-            "0.000282",
-            "0.000279",
-            "0.000280",
-            "0.000281",
-            "0.000281",
-            "0.000286",
-            "0.000278",
-            "0.000278",
-            "0.000281",
-            "0.000276",
-            "0.000284",
-            "0.000281",
-            "0.000276",
-            "0.000276",
-            "0.000279",
-            "0.000283",
-            "0.000282",
-            "0.000278",
-            "0.000281",
-            "0.000284",
-            "0.000279",
-            "0.000276",
-            "0.000278",
-            "0.000283",
-            "0.000282",
-            "0.000276",
-            "0.000281",
-            "0.000279",
-            "0.000276",
-            "0.000277",
-            "0.000283",
-            "0.000279",
-            "0.000281",
-            "0.000283",
-            "0.000279",
-            "0.000282",
-            "0.000283",
-            "0.000278",
-            "0.000281",
-            "0.000282",
-            "0.000278",
-            "0.000276",
-            "0.000281",
-            "0.000278",
-            "0.000276",
-            "0.000282",
-            "0.000281",
-            "0.000282",
-            "0.000280",
-            "0.000281",
-            "0.000282",
-            "0.000280",
-            "0.000282",
-            "0.000280",
-            "0.000280",
-            "0.000282",
-            "0.000278",
-            "0.000284",
-            "0.000290",
-            "0.000282",
-            "0.000281",
-            "0.000281",
-            "0.000281",
-            "0.000278"
-        ],
-        "iqr": "0.000004",
-        "iterations": 100,
-        "max": "0.000290",
-        "mean": "0.000280",
-        "median": "0.000281",
-        "min": "0.000275",
-        "q1": "0.000278",
-        "q3": "0.000282",
-        "stdev": "0.000003",
-        "time_unit": "s",
-        "times": [],
-        "unit": "s"
-    },
-    "tags": {
-        "name": "nlike_utf8 scalar starts with",
-        "suite": "nlike_utf8 scalar starts with"
-    },
-    "timestamp": "2022-02-09T02:33:26.792404+00:00"
-}
-```
-
-## Debug with test benchmark
-
-```
-(conbench) $ cd ~/arrow-rs/conbench/
-(conbench) $ conbench test --iterations=3
-
-Benchmark result:
-{
-    "batch_id": "f4235d547e9d4f94925b54692e625d7d",
-    "context": {
-        "benchmark_language": "Python"
-    },
-    "github": {
-        "commit": "35e16be01e680e9381b2d1393c2e3f8e7acb7b13",
-        "repository": "https://github.com/dianaclarke/arrow-rs"
-    },
-    "info": {
-        "benchmark_language_version": "Python 3.9.7"
-    },
-    "machine_info": {
-        "architecture_name": "x86_64",
-        "cpu_core_count": "8",
-        "cpu_frequency_max_hz": "2400000000",
-        "cpu_l1d_cache_bytes": "65536",
-        "cpu_l1i_cache_bytes": "131072",
-        "cpu_l2_cache_bytes": "4194304",
-        "cpu_l3_cache_bytes": "0",
-        "cpu_model_name": "Apple M1",
-        "cpu_thread_count": "8",
-        "gpu_count": "0",
-        "gpu_product_names": [],
-        "kernel_name": "20.6.0",
-        "memory_bytes": "17179869184",
-        "name": "diana",
-        "os_name": "macOS",
-        "os_version": "10.16"
-    },
-    "run_id": "b2ca0d581cf14f21936276ff7ca5a940",
-    "stats": {
-        "data": [
-            "0.000002",
-            "0.000001",
-            "0.000001"
-        ],
-        "iqr": "0.000001",
-        "iterations": 3,
-        "max": "0.000002",
-        "mean": "0.000001",
-        "median": "0.000001",
-        "min": "0.000001",
-        "q1": "0.000001",
-        "q3": "0.000002",
-        "stdev": "0.000001",
-        "time_unit": "s",
-        "times": [],
-        "unit": "s"
-    },
-    "tags": {
-        "name": "test"
-    },
-    "timestamp": "2022-02-09T01:55:52.250727+00:00"
-}
-```
diff --git a/conbench/_criterion.py b/conbench/_criterion.py
deleted file mode 100644
index 168a1b9b6..000000000
--- a/conbench/_criterion.py
+++ /dev/null
@@ -1,98 +0,0 @@
-# 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 collections
-import csv
-import os
-import pathlib
-import subprocess
-
-import conbench.runner
-from conbench.machine_info import github_info
-
-
-def _result_in_seconds(row):
-    # sample_measured_value - The value of the measurement for this sample.
-    # Note that this is the measured value for the whole sample, not the
-    # time-per-iteration To calculate the time-per-iteration, use
-    # sample_measured_value/iteration_count
-    # -- https://bheisler.github.io/criterion.rs/book/user_guide/csv_output.html
-    count = int(row["iteration_count"])
-    sample = float(row["sample_measured_value"])
-    return sample / count / 10**9
-
-
-def _parse_benchmark_group(row):
-    parts = row["group"].split(",")
-    if len(parts) > 1:
-        suite, name = parts[0], ",".join(parts[1:])
-    else:
-        suite, name = row["group"], row["group"]
-    return suite.strip(), name.strip()
-
-
-def _read_results(src_dir):
-    results = collections.defaultdict(lambda: collections.defaultdict(list))
-    path = pathlib.Path(os.path.join(src_dir, "target", "criterion"))
-    for path in list(path.glob("**/new/raw.csv")):
-        with open(path) as csv_file:
-            reader = csv.DictReader(csv_file)
-            for row in reader:
-                suite, name = _parse_benchmark_group(row)
-                results[suite][name].append(_result_in_seconds(row))
-    return results
-
-
-def _execute_command(command):
-    try:
-        print(command)
-        result = subprocess.run(command, capture_output=True, check=True)
-    except subprocess.CalledProcessError as e:
-        print(e.stderr.decode("utf-8"))
-        raise e
-    return result.stdout.decode("utf-8"), result.stderr.decode("utf-8")
-
-
-class CriterionBenchmark(conbench.runner.Benchmark):
-    external = True
-
-    def run(self, **kwargs):
-        src_dir = os.path.join(os.getcwd(), "..")
-        self._cargo_bench(src_dir)
-        results = _read_results(src_dir)
-        for suite in results:
-            self.conbench.mark_new_batch()
-            for name, data in results[suite].items():
-                yield self._record_result(suite, name, data, kwargs)
-
-    def _cargo_bench(self, src_dir):
-        os.chdir(src_dir)
-        _execute_command(["cargo", "bench"])
-
-    def _record_result(self, suite, name, data, options):
-        tags = {"suite": suite}
-        result = {"data": data, "unit": "s"}
-        context = {"benchmark_language": "Rust"}
-        github = github_info()
-        return self.conbench.record(
-            result,
-            name,
-            tags=tags,
-            context=context,
-            github=github,
-            options=options,
-        )
diff --git a/conbench/benchmarks.json b/conbench/benchmarks.json
deleted file mode 100644
index 3e1f33a7d..000000000
--- a/conbench/benchmarks.json
+++ /dev/null
@@ -1,8 +0,0 @@
-[
-  {
-    "command": "arrow-rs",
-    "flags": {
-      "language": "Rust"
-    }
-  }
-]
diff --git a/conbench/benchmarks.py b/conbench/benchmarks.py
deleted file mode 100644
index bc4c1796b..000000000
--- a/conbench/benchmarks.py
+++ /dev/null
@@ -1,41 +0,0 @@
-# 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 conbench.runner
-
-import _criterion
-
-
-@conbench.runner.register_benchmark
-class TestBenchmark(conbench.runner.Benchmark):
-    name = "test"
-
-    def run(self, **kwargs):
-        yield self.conbench.benchmark(
-            self._f(),
-            self.name,
-            options=kwargs,
-        )
-
-    def _f(self):
-        return lambda: 1 + 1
-
-
-@conbench.runner.register_benchmark
-class CargoBenchmarks(_criterion.CriterionBenchmark):
-    name = "arrow-rs"
-    description = "Run Arrow Rust micro benchmarks."
diff --git a/conbench/requirements.txt b/conbench/requirements.txt
deleted file mode 100644
index a877c7b44..000000000
--- a/conbench/requirements.txt
+++ /dev/null
@@ -1,1 +0,0 @@
-conbench
diff --git a/dev/release/rat_exclude_files.txt b/dev/release/rat_exclude_files.txt
index 6f9d1b5f3..d08a0ea8c 100644
--- a/dev/release/rat_exclude_files.txt
+++ b/dev/release/rat_exclude_files.txt
@@ -14,11 +14,6 @@ parquet_derive/test/dependency/*
 Cargo.lock
 filtered_rat.txt
 rat.txt
-conbench/benchmarks.json
-conbench/requirements.txt
-conbench/requirements-test.txt
-conbench/.flake8
-conbench/.isort.cfg
 # auto-generated
 arrow-flight/src/arrow.flight.protocol.rs
 arrow-flight/src/sql/arrow.flight.protocol.sql.rs
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index 574345c38..75acb73e5 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -402,7 +402,7 @@ impl FromStr for AmazonS3ConfigKey {
             "aws_sse_customer_key_base64" => Ok(Self::Encryption(
                 S3EncryptionConfigKey::CustomerEncryptionKey,
             )),
-            _ => match s.parse() {
+            _ => match s.strip_prefix("aws_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
             },
@@ -1455,4 +1455,17 @@ mod tests {
             assert_eq!(parse_bucket_az(bucket), expected)
         }
     }
+
+    #[test]
+    fn aws_test_client_opts() {
+        let key = "AWS_PROXY_URL";
+        if let Ok(config_key) = key.to_ascii_lowercase().parse() {
+            assert_eq!(
+                AmazonS3ConfigKey::Client(ClientConfigKey::ProxyUrl),
+                config_key
+            );
+        } else {
+            panic!("{} not propagated as ClientConfigKey", key);
+        }
+    }
 }
diff --git a/object_store/src/azure/builder.rs b/object_store/src/azure/builder.rs
index c0c4e8983..0208073e8 100644
--- a/object_store/src/azure/builder.rs
+++ b/object_store/src/azure/builder.rs
@@ -408,7 +408,7 @@ impl FromStr for AzureConfigKey {
             "azure_disable_tagging" | "disable_tagging" => Ok(Self::DisableTagging),
             // Backwards compatibility
             "azure_allow_http" => Ok(Self::Client(ClientConfigKey::AllowHttp)),
-            _ => match s.parse() {
+            _ => match s.strip_prefix("azure_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
             },
@@ -1103,4 +1103,17 @@ mod tests {
         let pairs = split_sas(raw_sas).unwrap();
         assert_eq!(expected, pairs);
     }
+
+    #[test]
+    fn azure_test_client_opts() {
+        let key = "AZURE_PROXY_URL";
+        if let Ok(config_key) = key.to_ascii_lowercase().parse() {
+            assert_eq!(
+                AzureConfigKey::Client(ClientConfigKey::ProxyUrl),
+                config_key
+            );
+        } else {
+            panic!("{} not propagated as ClientConfigKey", key);
+        }
+    }
 }
diff --git a/object_store/src/client/mod.rs b/object_store/src/client/mod.rs
index c45833b89..9a3b70554 100644
--- a/object_store/src/client/mod.rs
+++ b/object_store/src/client/mod.rs
@@ -157,6 +157,8 @@ impl FromStr for ClientConfigKey {
             "pool_idle_timeout" => Ok(Self::PoolIdleTimeout),
             "pool_max_idle_per_host" => Ok(Self::PoolMaxIdlePerHost),
             "proxy_url" => Ok(Self::ProxyUrl),
+            "proxy_ca_certificate" => Ok(Self::ProxyCaCertificate),
+            "proxy_excludes" => Ok(Self::ProxyExcludes),
             "timeout" => Ok(Self::Timeout),
             "user_agent" => Ok(Self::UserAgent),
             _ => Err(super::Error::UnknownConfigurationKey {
diff --git a/object_store/src/delimited.rs b/object_store/src/delimited.rs
index 4f25c9d6d..96f88bf41 100644
--- a/object_store/src/delimited.rs
+++ b/object_store/src/delimited.rs
@@ -126,7 +126,7 @@ impl LineDelimiter {
     fn finish(&mut self) -> Result<bool> {
         if !self.remainder.is_empty() {
             ensure!(!self.is_quote, UnterminatedStringSnafu);
-            ensure!(!self.is_quote, TrailingEscapeSnafu);
+            ensure!(!self.is_escape, TrailingEscapeSnafu);
 
             self.complete
                 .push_back(Bytes::from(std::mem::take(&mut self.remainder)))
diff --git a/object_store/src/gcp/builder.rs b/object_store/src/gcp/builder.rs
index 82dab1443..26cc8211d 100644
--- a/object_store/src/gcp/builder.rs
+++ b/object_store/src/gcp/builder.rs
@@ -185,7 +185,7 @@ impl FromStr for GoogleConfigKey {
             "google_service_account_key" | "service_account_key" => Ok(Self::ServiceAccountKey),
             "google_bucket" | "google_bucket_name" | "bucket" | "bucket_name" => Ok(Self::Bucket),
             "google_application_credentials" => Ok(Self::ApplicationCredentials),
-            _ => match s.parse() {
+            _ => match s.strip_prefix("google_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
             },
@@ -671,4 +671,17 @@ mod tests {
             google_bucket_name
         );
     }
+
+    #[test]
+    fn gcp_test_client_opts() {
+        let key = "GOOGLE_PROXY_URL";
+        if let Ok(config_key) = key.to_ascii_lowercase().parse() {
+            assert_eq!(
+                GoogleConfigKey::Client(ClientConfigKey::ProxyUrl),
+                config_key
+            );
+        } else {
+            panic!("{} not propagated as ClientConfigKey", key);
+        }
+    }
 }
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 4b43f0cde..8820983b2 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -911,7 +911,7 @@ pub struct ObjectMeta {
 }
 
 /// Options for a get request, such as range
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct GetOptions {
     /// Request will succeed if the `ObjectMeta::e_tag` matches
     /// otherwise returning [`Error::Precondition`]
diff --git a/parquet/src/arrow/array_reader/empty_array.rs b/parquet/src/arrow/array_reader/empty_array.rs
index 51673f2f8..9aba08b4e 100644
--- a/parquet/src/arrow/array_reader/empty_array.rs
+++ b/parquet/src/arrow/array_reader/empty_array.rs
@@ -17,9 +17,9 @@
 
 use crate::arrow::array_reader::ArrayReader;
 use crate::errors::Result;
-use arrow_schema::{DataType as ArrowType, Fields};
 use arrow_array::{ArrayRef, StructArray};
 use arrow_data::ArrayDataBuilder;
+use arrow_schema::{DataType as ArrowType, Fields};
 use std::any::Any;
 use std::sync::Arc;
 
diff --git a/parquet/src/arrow/array_reader/fixed_size_list_array.rs b/parquet/src/arrow/array_reader/fixed_size_list_array.rs
index 4cf68a066..75099d018 100644
--- a/parquet/src/arrow/array_reader/fixed_size_list_array.rs
+++ b/parquet/src/arrow/array_reader/fixed_size_list_array.rs
@@ -138,7 +138,7 @@ impl ArrayReader for FixedSizeListArrayReader {
                             "Encountered misaligned row with length {} (expected length {})",
                             row_len,
                             self.fixed_size
-                        ))
+                        ));
                     }
                     row_len = 0;
 
@@ -226,9 +226,7 @@ mod tests {
     use super::*;
     use crate::arrow::{
         array_reader::{test_util::InMemoryArrayReader, ListArrayReader},
-        arrow_reader::{
-            ArrowReaderBuilder, ArrowReaderOptions, ParquetRecordBatchReader,
-        },
+        arrow_reader::{ArrowReaderBuilder, ArrowReaderOptions, ParquetRecordBatchReader},
         ArrowWriter,
     };
     use arrow::datatypes::{Field, Int32Type};
@@ -279,10 +277,7 @@ mod tests {
         let mut list_array_reader = FixedSizeListArrayReader::new(
             Box::new(item_array_reader),
             3,
-            ArrowType::FixedSizeList(
-                Arc::new(Field::new("item", ArrowType::Int32, true)),
-                3,
-            ),
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", ArrowType::Int32, true)), 3),
             2,
             1,
             true,
@@ -328,10 +323,7 @@ mod tests {
         let mut list_array_reader = FixedSizeListArrayReader::new(
             Box::new(item_array_reader),
             2,
-            ArrowType::FixedSizeList(
-                Arc::new(Field::new("item", ArrowType::Int32, true)),
-                2,
-            ),
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", ArrowType::Int32, true)), 2),
             1,
             1,
             false,
@@ -354,14 +346,10 @@ mod tests {
         //   [[4, 5]],
         //   [[null, null]],
         // ]
-        let l2_type = ArrowType::FixedSizeList(
-            Arc::new(Field::new("item", ArrowType::Int32, true)),
-            2,
-        );
-        let l1_type = ArrowType::FixedSizeList(
-            Arc::new(Field::new("item", l2_type.clone(), false)),
-            1,
-        );
+        let l2_type =
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", ArrowType::Int32, true)), 2);
+        let l1_type =
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", l2_type.clone(), false)), 1);
 
         let array = PrimitiveArray::<Int32Type>::from(vec![
             None,
@@ -413,14 +401,8 @@ mod tests {
             Some(vec![0, 0, 2, 0, 2, 0, 0, 2, 0, 2]),
         );
 
-        let l2 = FixedSizeListArrayReader::new(
-            Box::new(item_array_reader),
-            2,
-            l2_type,
-            4,
-            2,
-            false,
-        );
+        let l2 =
+            FixedSizeListArrayReader::new(Box::new(item_array_reader), 2, l2_type, 4, 2, false);
         let mut l1 = FixedSizeListArrayReader::new(Box::new(l2), 1, l1_type, 3, 1, true);
 
         let expected_1 = expected.slice(0, 2);
@@ -454,10 +436,7 @@ mod tests {
         let mut list_array_reader = FixedSizeListArrayReader::new(
             Box::new(item_array_reader),
             0,
-            ArrowType::FixedSizeList(
-                Arc::new(Field::new("item", ArrowType::Int32, true)),
-                0,
-            ),
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", ArrowType::Int32, true)), 0),
             2,
             1,
             true,
@@ -473,8 +452,7 @@ mod tests {
     #[test]
     fn test_nested_var_list() {
         // [[[1, null, 3], null], [[4], []], [[5, 6], [null, null]], null]
-        let mut builder =
-            FixedSizeListBuilder::new(ListBuilder::new(Int32Builder::new()), 2);
+        let mut builder = FixedSizeListBuilder::new(ListBuilder::new(Int32Builder::new()), 2);
         builder.values().append_value([Some(1), None, Some(3)]);
         builder.values().append_null();
         builder.append(true);
@@ -503,12 +481,9 @@ mod tests {
             None,
         ]));
 
-        let inner_type =
-            ArrowType::List(Arc::new(Field::new("item", ArrowType::Int32, true)));
-        let list_type = ArrowType::FixedSizeList(
-            Arc::new(Field::new("item", inner_type.clone(), true)),
-            2,
-        );
+        let inner_type = ArrowType::List(Arc::new(Field::new("item", ArrowType::Int32, true)));
+        let list_type =
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", inner_type.clone(), true)), 2);
 
         let item_array_reader = InMemoryArrayReader::new(
             ArrowType::Int32,
@@ -517,22 +492,11 @@ mod tests {
             Some(vec![0, 2, 2, 1, 0, 1, 0, 2, 1, 2, 0]),
         );
 
-        let inner_array_reader = ListArrayReader::<i32>::new(
-            Box::new(item_array_reader),
-            inner_type,
-            4,
-            2,
-            true,
-        );
+        let inner_array_reader =
+            ListArrayReader::<i32>::new(Box::new(item_array_reader), inner_type, 4, 2, true);
 
-        let mut list_array_reader = FixedSizeListArrayReader::new(
-            Box::new(inner_array_reader),
-            2,
-            list_type,
-            2,
-            1,
-            true,
-        );
+        let mut list_array_reader =
+            FixedSizeListArrayReader::new(Box::new(inner_array_reader), 2, list_type, 2, 1, true);
         let actual = list_array_reader.next_batch(1024).unwrap();
         let actual = actual
             .as_any()
@@ -564,21 +528,13 @@ mod tests {
         );
 
         // [null, 2, 3, null, 5]
-        let primitive = PrimitiveArray::<Int32Type>::from_iter(vec![
-            None,
-            Some(2),
-            Some(3),
-            None,
-            Some(5),
-        ]);
+        let primitive =
+            PrimitiveArray::<Int32Type>::from_iter(vec![None, Some(2), Some(3), None, Some(5)]);
 
         let schema = Arc::new(Schema::new(vec![
             Field::new(
                 "list",
-                ArrowType::FixedSizeList(
-                    Arc::new(Field::new("item", ArrowType::Int32, true)),
-                    4,
-                ),
+                ArrowType::FixedSizeList(Arc::new(Field::new("item", ArrowType::Int32, true)), 4),
                 true,
             ),
             Field::new("primitive", ArrowType::Int32, true),
@@ -643,10 +599,7 @@ mod tests {
 
         let schema = Arc::new(Schema::new(vec![Field::new(
             "list",
-            ArrowType::FixedSizeList(
-                Arc::new(Field::new("item", ArrowType::Int32, true)),
-                4,
-            ),
+            ArrowType::FixedSizeList(Arc::new(Field::new("item", ArrowType::Int32, true)), 4),
             true,
         )]));
 
diff --git a/parquet/src/arrow/array_reader/list_array.rs b/parquet/src/arrow/array_reader/list_array.rs
index e1752f30c..ebff3286b 100644
--- a/parquet/src/arrow/array_reader/list_array.rs
+++ b/parquet/src/arrow/array_reader/list_array.rs
@@ -125,8 +125,7 @@ impl<OffsetSize: OffsetSizeTrait> ArrayReader for ListArrayReader<OffsetSize> {
         // lists, and for consistency we do the same for nulls.
 
         // The output offsets for the computed ListArray
-        let mut list_offsets: Vec<OffsetSize> =
-            Vec::with_capacity(next_batch_array.len() + 1);
+        let mut list_offsets: Vec<OffsetSize> = Vec::with_capacity(next_batch_array.len() + 1);
 
         // The validity mask of the computed ListArray if nullable
         let mut validity = self
@@ -270,9 +269,7 @@ mod tests {
         GenericListArray::<OffsetSize>::DATA_TYPE_CONSTRUCTOR(field)
     }
 
-    fn downcast<OffsetSize: OffsetSizeTrait>(
-        array: &ArrayRef,
-    ) -> &'_ GenericListArray<OffsetSize> {
+    fn downcast<OffsetSize: OffsetSizeTrait>(array: &ArrayRef) -> &'_ GenericListArray<OffsetSize> {
         array
             .as_any()
             .downcast_ref::<GenericListArray<OffsetSize>>()
@@ -383,18 +380,12 @@ mod tests {
             Some(vec![0, 3, 2, 2, 2, 1, 1, 1, 1, 3, 3, 2, 3, 3, 2, 0, 0, 0]),
         );
 
-        let l3 = ListArrayReader::<OffsetSize>::new(
-            Box::new(item_array_reader),
-            l3_type,
-            5,
-            3,
-            true,
-        );
+        let l3 =
+            ListArrayReader::<OffsetSize>::new(Box::new(item_array_reader), l3_type, 5, 3, true);
 
         let l2 = ListArrayReader::<OffsetSize>::new(Box::new(l3), l2_type, 3, 2, false);
 
-        let mut l1 =
-            ListArrayReader::<OffsetSize>::new(Box::new(l2), l1_type, 2, 1, true);
+        let mut l1 = ListArrayReader::<OffsetSize>::new(Box::new(l2), l1_type, 2, 1, true);
 
         let expected_1 = expected.slice(0, 2);
         let expected_2 = expected.slice(2, 2);
@@ -560,8 +551,7 @@ mod tests {
         .unwrap();
         writer.close().unwrap();
 
-        let file_reader: Arc<dyn FileReader> =
-            Arc::new(SerializedFileReader::new(file).unwrap());
+        let file_reader: Arc<dyn FileReader> = Arc::new(SerializedFileReader::new(file).unwrap());
 
         let file_metadata = file_reader.metadata().file_metadata();
         let schema = file_metadata.schema_descr();
@@ -573,8 +563,7 @@ mod tests {
         )
         .unwrap();
 
-        let mut array_reader =
-            build_array_reader(fields.as_ref(), &mask, &file_reader).unwrap();
+        let mut array_reader = build_array_reader(fields.as_ref(), &mask, &file_reader).unwrap();
 
         let batch = array_reader.next_batch(100).unwrap();
         assert_eq!(batch.data_type(), array_reader.get_data_type());
@@ -584,9 +573,7 @@ mod tests {
                 "table_info",
                 ArrowType::List(Arc::new(Field::new(
                     "table_info",
-                    ArrowType::Struct(
-                        vec![Field::new("name", ArrowType::Binary, false)].into()
-                    ),
+                    ArrowType::Struct(vec![Field::new("name", ArrowType::Binary, false)].into()),
                     false
                 ))),
                 false
diff --git a/parquet/src/arrow/array_reader/map_array.rs b/parquet/src/arrow/array_reader/map_array.rs
index 9bfc04732..4bdec602b 100644
--- a/parquet/src/arrow/array_reader/map_array.rs
+++ b/parquet/src/arrow/array_reader/map_array.rs
@@ -184,21 +184,19 @@ mod tests {
         map_builder.append(true).expect("adding map entry");
 
         // Create record batch
-        let batch =
-            RecordBatch::try_new(Arc::new(schema), vec![Arc::new(map_builder.finish())])
-                .expect("create record batch");
+        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(map_builder.finish())])
+            .expect("create record batch");
 
         // Write record batch to file
         let mut buffer = Vec::with_capacity(1024);
-        let mut writer = ArrowWriter::try_new(&mut buffer, batch.schema(), None)
-            .expect("creat file writer");
+        let mut writer =
+            ArrowWriter::try_new(&mut buffer, batch.schema(), None).expect("creat file writer");
         writer.write(&batch).expect("writing file");
         writer.close().expect("close writer");
 
         // Read file
         let reader = Bytes::from(buffer);
-        let record_batch_reader =
-            ParquetRecordBatchReader::try_new(reader, 1024).unwrap();
+        let record_batch_reader = ParquetRecordBatchReader::try_new(reader, 1024).unwrap();
         for maybe_record_batch in record_batch_reader {
             let record_batch = maybe_record_batch.expect("Getting current batch");
             let col = record_batch.column(0);
diff --git a/parquet/src/arrow/array_reader/struct_array.rs b/parquet/src/arrow/array_reader/struct_array.rs
index 4af194774..fb2f2f892 100644
--- a/parquet/src/arrow/array_reader/struct_array.rs
+++ b/parquet/src/arrow/array_reader/struct_array.rs
@@ -112,10 +112,10 @@ impl ArrayReader for StructArrayReader {
             .collect::<Result<Vec<_>>>()?;
 
         // check that array child data has same size
-        let children_array_len =
-            children_array.first().map(|arr| arr.len()).ok_or_else(|| {
-                general_err!("Struct array reader should have at least one child!")
-            })?;
+        let children_array_len = children_array
+            .first()
+            .map(|arr| arr.len())
+            .ok_or_else(|| general_err!("Struct array reader should have at least one child!"))?;
 
         let all_children_len_eq = children_array
             .iter()
@@ -169,8 +169,7 @@ impl ArrayReader for StructArrayReader {
                 return Err(general_err!("Failed to decode level data for struct array"));
             }
 
-            array_data_builder =
-                array_data_builder.null_bit_buffer(Some(bitmap_builder.into()));
+            array_data_builder = array_data_builder.null_bit_buffer(Some(bitmap_builder.into()));
         }
 
         let array_data = unsafe { array_data_builder.build_unchecked() };
@@ -282,13 +281,12 @@ mod tests {
         //    null,
         // ]
 
-        let expected_l =
-            Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
-                Some(vec![Some(1), Some(2), None]),
-                Some(vec![]),
-                None,
-                None,
-            ]));
+        let expected_l = Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
+            Some(vec![Some(1), Some(2), None]),
+            Some(vec![]),
+            None,
+            None,
+        ]));
 
         let validity = Buffer::from([0b00000111]);
         let struct_fields = vec![(
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs b/parquet/src/arrow/arrow_reader/statistics.rs
index 602a9ad5e..4c0dac05c 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -568,13 +568,9 @@ macro_rules! make_data_page_stats_iterator {
                 let next = self.iter.next();
                 match next {
                     Some((len, index)) => match index {
-                        $index_type(native_index) => Some(
-                            native_index
-                                .indexes
-                                .iter()
-                                .map(|x| $func(x))
-                                .collect::<Vec<_>>(),
-                        ),
+                        $index_type(native_index) => {
+                            Some(native_index.indexes.iter().map($func).collect::<Vec<_>>())
+                        }
                         // No matching `Index` found;
                         // thus no statistics that can be extracted.
                         // We return vec![None; len] to effectively
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 5c5f28ac0..3ec7a3dfe 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -3082,8 +3082,8 @@ mod tests {
             let min = byte_array_stats.min_opt().unwrap();
             let max = byte_array_stats.max_opt().unwrap();
 
-            assert_eq!(min.as_bytes(), &[b'a']);
-            assert_eq!(max.as_bytes(), &[b'd']);
+            assert_eq!(min.as_bytes(), b"a");
+            assert_eq!(max.as_bytes(), b"d");
         } else {
             panic!("expecting Statistics::ByteArray");
         }
@@ -3154,8 +3154,8 @@ mod tests {
             let min = byte_array_stats.min_opt().unwrap();
             let max = byte_array_stats.max_opt().unwrap();
 
-            assert_eq!(min.as_bytes(), &[b'a']);
-            assert_eq!(max.as_bytes(), &[b'd']);
+            assert_eq!(min.as_bytes(), b"a");
+            assert_eq!(max.as_bytes(), b"d");
         } else {
             panic!("expecting Statistics::ByteArray");
         }
diff --git a/parquet/src/arrow/schema/complex.rs b/parquet/src/arrow/schema/complex.rs
index a5f6d242e..e487feabb 100644
--- a/parquet/src/arrow/schema/complex.rs
+++ b/parquet/src/arrow/schema/complex.rs
@@ -41,7 +41,7 @@ pub struct ParquetField {
     /// i.e. guaranteed to be > 0 for an element of list type
     pub rep_level: i16,
     /// The level at which this field is fully defined,
-    /// i.e. guaranteed to be > 0 for a nullable type or child of a 
+    /// i.e. guaranteed to be > 0 for a nullable type or child of a
     /// nullable type
     pub def_level: i16,
     /// Whether this field is nullable
@@ -64,11 +64,7 @@ impl ParquetField {
             rep_level: self.rep_level,
             def_level: self.def_level,
             nullable: false,
-            arrow_type: DataType::List(Arc::new(Field::new(
-                name,
-                self.arrow_type.clone(),
-                false,
-            ))),
+            arrow_type: DataType::List(Arc::new(Field::new(name, self.arrow_type.clone(), false))),
             field_type: ParquetFieldType::Group {
                 children: vec![self],
             },
@@ -289,7 +285,7 @@ impl Visitor {
 
         match map_key.get_basic_info().repetition() {
             Repetition::REPEATED => {
-            return Err(arrow_err!("Map keys cannot be repeated"));
+                return Err(arrow_err!("Map keys cannot be repeated"));
             }
             Repetition::REQUIRED | Repetition::OPTIONAL => {
                 // Relaxed check for having repetition REQUIRED as there exists
@@ -317,10 +313,7 @@ impl Visitor {
                     (Some(field), Some(&*fields[0]), Some(&*fields[1]), *sorted)
                 }
                 d => {
-                    return Err(arrow_err!(
-                        "Map data type should contain struct got {}",
-                        d
-                    ));
+                    return Err(arrow_err!("Map data type should contain struct got {}", d));
                 }
             },
             Some(d) => {
@@ -416,9 +409,7 @@ impl Visitor {
         let (def_level, nullable) = match list_type.get_basic_info().repetition() {
             Repetition::REQUIRED => (context.def_level, false),
             Repetition::OPTIONAL => (context.def_level + 1, true),
-            Repetition::REPEATED => {
-                return Err(arrow_err!("List type cannot be repeated"))
-            }
+            Repetition::REPEATED => return Err(arrow_err!("List type cannot be repeated")),
         };
 
         let arrow_field = match &context.data_type {
@@ -542,11 +533,7 @@ impl Visitor {
 ///
 /// The resulting [`Field`] will have the type dictated by `field`, a name
 /// dictated by the `parquet_type`, and any metadata from `arrow_hint`
-fn convert_field(
-    parquet_type: &Type,
-    field: &ParquetField,
-    arrow_hint: Option<&Field>,
-) -> Field {
+fn convert_field(parquet_type: &Type, field: &ParquetField, arrow_hint: Option<&Field>) -> Field {
     let name = parquet_type.name();
     let data_type = field.arrow_type.clone();
     let nullable = field.nullable;
@@ -568,11 +555,14 @@ fn convert_field(
             let basic_info = parquet_type.get_basic_info();
             if basic_info.has_id() {
                 let mut meta = HashMap::with_capacity(1);
-                meta.insert(PARQUET_FIELD_ID_META_KEY.to_string(), basic_info.id().to_string());
+                meta.insert(
+                    PARQUET_FIELD_ID_META_KEY.to_string(),
+                    basic_info.id().to_string(),
+                );
                 ret.set_metadata(meta);
             }
             ret
-        },
+        }
     }
 }
 
diff --git a/parquet/src/arrow/schema/mod.rs b/parquet/src/arrow/schema/mod.rs
index a3528b6c8..fab896695 100644
--- a/parquet/src/arrow/schema/mod.rs
+++ b/parquet/src/arrow/schema/mod.rs
@@ -1483,11 +1483,25 @@ mod tests {
             ),
             Field::new("date", DataType::Date32, true),
             Field::new("time_milli", DataType::Time32(TimeUnit::Millisecond), true),
-            Field::new("time_milli_utc", DataType::Time32(TimeUnit::Millisecond), true)
-                .with_metadata(HashMap::from_iter(vec![("adjusted_to_utc".to_string(), "".to_string())])),
+            Field::new(
+                "time_milli_utc",
+                DataType::Time32(TimeUnit::Millisecond),
+                true,
+            )
+            .with_metadata(HashMap::from_iter(vec![(
+                "adjusted_to_utc".to_string(),
+                "".to_string(),
+            )])),
             Field::new("time_micro", DataType::Time64(TimeUnit::Microsecond), true),
-            Field::new("time_micro_utc", DataType::Time64(TimeUnit::Microsecond), true)
-                .with_metadata(HashMap::from_iter(vec![("adjusted_to_utc".to_string(), "".to_string())])),
+            Field::new(
+                "time_micro_utc",
+                DataType::Time64(TimeUnit::Microsecond),
+                true,
+            )
+            .with_metadata(HashMap::from_iter(vec![(
+                "adjusted_to_utc".to_string(),
+                "".to_string(),
+            )])),
             Field::new(
                 "ts_milli",
                 DataType::Timestamp(TimeUnit::Millisecond, None),
diff --git a/parquet/src/compression.rs b/parquet/src/compression.rs
index 10560210e..edf675f13 100644
--- a/parquet/src/compression.rs
+++ b/parquet/src/compression.rs
@@ -150,35 +150,47 @@ pub fn create_codec(codec: CodecType, _options: &CodecOptions) -> Result<Option<
         CodecType::BROTLI(level) => {
             #[cfg(any(feature = "brotli", test))]
             return Ok(Some(Box::new(BrotliCodec::new(level))));
-            Err(ParquetError::General("Disabled feature at compile time: brotli".into()))
-        },
+            Err(ParquetError::General(
+                "Disabled feature at compile time: brotli".into(),
+            ))
+        }
         CodecType::GZIP(level) => {
             #[cfg(any(feature = "flate2", test))]
             return Ok(Some(Box::new(GZipCodec::new(level))));
-            Err(ParquetError::General("Disabled feature at compile time: flate2".into()))
-        },
+            Err(ParquetError::General(
+                "Disabled feature at compile time: flate2".into(),
+            ))
+        }
         CodecType::SNAPPY => {
             #[cfg(any(feature = "snap", test))]
             return Ok(Some(Box::new(SnappyCodec::new())));
-            Err(ParquetError::General("Disabled feature at compile time: snap".into()))
-        },
+            Err(ParquetError::General(
+                "Disabled feature at compile time: snap".into(),
+            ))
+        }
         CodecType::LZ4 => {
             #[cfg(any(feature = "lz4", test))]
             return Ok(Some(Box::new(LZ4HadoopCodec::new(
                 _options.backward_compatible_lz4,
             ))));
-            Err(ParquetError::General("Disabled feature at compile time: lz4".into()))
-        },
+            Err(ParquetError::General(
+                "Disabled feature at compile time: lz4".into(),
+            ))
+        }
         CodecType::ZSTD(level) => {
             #[cfg(any(feature = "zstd", test))]
             return Ok(Some(Box::new(ZSTDCodec::new(level))));
-            Err(ParquetError::General("Disabled feature at compile time: zstd".into()))
-        },
+            Err(ParquetError::General(
+                "Disabled feature at compile time: zstd".into(),
+            ))
+        }
         CodecType::LZ4_RAW => {
             #[cfg(any(feature = "lz4", test))]
             return Ok(Some(Box::new(LZ4RawCodec::new())));
-            Err(ParquetError::General("Disabled feature at compile time: lz4".into()))
-        },
+            Err(ParquetError::General(
+                "Disabled feature at compile time: lz4".into(),
+            ))
+        }
         CodecType::UNCOMPRESSED => Ok(None),
         _ => Err(nyi_err!("The codec type {} is not supported yet", codec)),
     }
diff --git a/parquet/src/data_type.rs b/parquet/src/data_type.rs
index 324e1c379..a3bcfd167 100644
--- a/parquet/src/data_type.rs
+++ b/parquet/src/data_type.rs
@@ -1298,11 +1298,8 @@ mod tests {
 
     #[test]
     fn test_byte_array_from() {
-        assert_eq!(
-            ByteArray::from(vec![b'A', b'B', b'C']).data(),
-            &[b'A', b'B', b'C']
-        );
-        assert_eq!(ByteArray::from("ABC").data(), &[b'A', b'B', b'C']);
+        assert_eq!(ByteArray::from(b"ABC".to_vec()).data(), b"ABC");
+        assert_eq!(ByteArray::from("ABC").data(), b"ABC");
         assert_eq!(
             ByteArray::from(Bytes::from(vec![1u8, 2u8, 3u8, 4u8, 5u8])).data(),
             &[1u8, 2u8, 3u8, 4u8, 5u8]
diff --git a/parquet/src/encodings/encoding/dict_encoder.rs b/parquet/src/encodings/encoding/dict_encoder.rs
index 98283b574..8efb84521 100644
--- a/parquet/src/encodings/encoding/dict_encoder.rs
+++ b/parquet/src/encodings/encoding/dict_encoder.rs
@@ -148,7 +148,6 @@ impl<T: DataType> DictEncoder<T> {
     fn bit_width(&self) -> u8 {
         num_required_bits(self.num_entries().saturating_sub(1) as u64)
     }
-
 }
 
 impl<T: DataType> Encoder<T> for DictEncoder<T> {
diff --git a/parquet/src/encodings/levels.rs b/parquet/src/encodings/levels.rs
index 8d4e48ba1..6f662b614 100644
--- a/parquet/src/encodings/levels.rs
+++ b/parquet/src/encodings/levels.rs
@@ -27,11 +27,7 @@ use crate::util::bit_util::{ceil, num_required_bits, BitWriter};
 /// repetition/definition level and number of total buffered values (includes null
 /// values).
 #[inline]
-pub fn max_buffer_size(
-    encoding: Encoding,
-    max_level: i16,
-    num_buffered_values: usize,
-) -> usize {
+pub fn max_buffer_size(encoding: Encoding, max_level: i16, num_buffered_values: usize) -> usize {
     let bit_width = num_required_bits(max_level as u64);
     match encoding {
         Encoding::RLE => RleEncoder::max_buffer_size(bit_width, num_buffered_values),
diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
index 97a122941..e1ca8cd74 100644
--- a/parquet/src/encodings/rle.rs
+++ b/parquet/src/encodings/rle.rs
@@ -199,13 +199,9 @@ impl RleEncoder {
     /// internal writer.
     #[inline]
     pub fn flush(&mut self) {
-        if self.bit_packed_count > 0
-            || self.repeat_count > 0
-            || self.num_buffered_values > 0
-        {
+        if self.bit_packed_count > 0 || self.repeat_count > 0 || self.num_buffered_values > 0 {
             let all_repeat = self.bit_packed_count == 0
-                && (self.repeat_count == self.num_buffered_values
-                    || self.num_buffered_values == 0);
+                && (self.repeat_count == self.num_buffered_values || self.num_buffered_values == 0);
             if self.repeat_count > 0 && all_repeat {
                 self.flush_rle_run();
             } else {
@@ -250,11 +246,8 @@ impl RleEncoder {
             // Write the indicator byte to the reserved position in `bit_writer`
             let num_groups = self.bit_packed_count / 8;
             let indicator_byte = ((num_groups << 1) | 1) as u8;
-            self.bit_writer.put_aligned_offset(
-                indicator_byte,
-                1,
-                self.indicator_byte_pos as usize,
-            );
+            self.bit_writer
+                .put_aligned_offset(indicator_byte, 1, self.indicator_byte_pos as usize);
             self.indicator_byte_pos = -1;
             self.bit_packed_count = 0;
         }
@@ -288,9 +281,7 @@ impl RleEncoder {
 
     /// return the estimated memory size of this encoder.
     pub(crate) fn estimated_memory_size(&self) -> usize {
-        self.bit_writer.estimated_memory_size()
-        + std::mem::size_of::<Self>()
-
+        self.bit_writer.estimated_memory_size() + std::mem::size_of::<Self>()
     }
 }
 
@@ -384,12 +375,10 @@ impl RleDecoder {
         let mut values_read = 0;
         while values_read < buffer.len() {
             if self.rle_left > 0 {
-                let num_values =
-                    cmp::min(buffer.len() - values_read, self.rle_left as usize);
+                let num_values = cmp::min(buffer.len() - values_read, self.rle_left as usize);
                 for i in 0..num_values {
-                    let repeated_value = T::try_from_le_slice(
-                        &self.current_value.as_mut().unwrap().to_ne_bytes(),
-                    )?;
+                    let repeated_value =
+                        T::try_from_le_slice(&self.current_value.as_mut().unwrap().to_ne_bytes())?;
                     buffer[values_read + i] = repeated_value;
                 }
                 self.rle_left -= num_values as u32;
@@ -397,8 +386,7 @@ impl RleDecoder {
             } else if self.bit_packed_left > 0 {
                 let mut num_values =
                     cmp::min(buffer.len() - values_read, self.bit_packed_left as usize);
-                let bit_reader =
-                    self.bit_reader.as_mut().expect("bit_reader should be set");
+                let bit_reader = self.bit_reader.as_mut().expect("bit_reader should be set");
 
                 num_values = bit_reader.get_batch::<T>(
                     &mut buffer[values_read..values_read + num_values],
@@ -424,15 +412,13 @@ impl RleDecoder {
         let mut values_skipped = 0;
         while values_skipped < num_values {
             if self.rle_left > 0 {
-                let num_values =
-                    cmp::min(num_values - values_skipped, self.rle_left as usize);
+                let num_values = cmp::min(num_values - values_skipped, self.rle_left as usize);
                 self.rle_left -= num_values as u32;
                 values_skipped += num_values;
             } else if self.bit_packed_left > 0 {
                 let mut num_values =
                     cmp::min(num_values - values_skipped, self.bit_packed_left as usize);
-                let bit_reader =
-                    self.bit_reader.as_mut().expect("bit_reader should be set");
+                let bit_reader = self.bit_reader.as_mut().expect("bit_reader should be set");
 
                 num_values = bit_reader.skip(num_values, self.bit_width as usize);
                 if num_values == 0 {
@@ -467,8 +453,7 @@ impl RleDecoder {
             let index_buf = self.index_buf.get_or_insert_with(|| Box::new([0; 1024]));
 
             if self.rle_left > 0 {
-                let num_values =
-                    cmp::min(max_values - values_read, self.rle_left as usize);
+                let num_values = cmp::min(max_values - values_read, self.rle_left as usize);
                 let dict_idx = self.current_value.unwrap() as usize;
                 for i in 0..num_values {
                     buffer[values_read + i].clone_from(&dict[dict_idx]);
@@ -476,8 +461,7 @@ impl RleDecoder {
                 self.rle_left -= num_values as u32;
                 values_read += num_values;
             } else if self.bit_packed_left > 0 {
-                let bit_reader =
-                    self.bit_reader.as_mut().expect("bit_reader should be set");
+                let bit_reader = self.bit_reader.as_mut().expect("bit_reader should be set");
 
                 loop {
                     let to_read = index_buf
@@ -489,10 +473,8 @@ impl RleDecoder {
                         break;
                     }
 
-                    let num_values = bit_reader.get_batch::<i32>(
-                        &mut index_buf[..to_read],
-                        self.bit_width as usize,
-                    );
+                    let num_values = bit_reader
+                        .get_batch::<i32>(&mut index_buf[..to_read], self.bit_width as usize);
                     if num_values == 0 {
                         // Handle writers which truncate the final block
                         self.bit_packed_left = 0;
@@ -708,14 +690,10 @@ mod tests {
         decoder.set_data(data.into());
         let mut buffer = vec![""; 12];
         let expected = vec![
-            "ddd", "eee", "fff", "ddd", "eee", "fff", "ddd", "eee", "fff", "eee", "fff",
-            "fff",
+            "ddd", "eee", "fff", "ddd", "eee", "fff", "ddd", "eee", "fff", "eee", "fff", "fff",
         ];
-        let result = decoder.get_batch_with_dict::<&str>(
-            dict.as_slice(),
-            buffer.as_mut_slice(),
-            12,
-        );
+        let result =
+            decoder.get_batch_with_dict::<&str>(dict.as_slice(), buffer.as_mut_slice(), 12);
         assert!(result.is_ok());
         assert_eq!(buffer, expected);
     }
@@ -1042,8 +1020,7 @@ mod tests {
         for _ in 0..niters {
             values.clear();
             let rng = thread_rng();
-            let seed_vec: Vec<u8> =
-                rng.sample_iter::<u8, _>(&Standard).take(seed_len).collect();
+            let seed_vec: Vec<u8> = rng.sample_iter::<u8, _>(&Standard).take(seed_len).collect();
             let mut seed = [0u8; 32];
             seed.copy_from_slice(&seed_vec[0..seed_len]);
             let mut gen = rand::rngs::StdRng::from_seed(seed);
diff --git a/parquet/src/file/metadata/writer.rs b/parquet/src/file/metadata/writer.rs
index dad960790..92ce60556 100644
--- a/parquet/src/file/metadata/writer.rs
+++ b/parquet/src/file/metadata/writer.rs
@@ -55,18 +55,14 @@ impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
         // write offset index to the file
         for (row_group_idx, row_group) in self.row_groups.iter_mut().enumerate() {
             for (column_idx, column_metadata) in row_group.columns.iter_mut().enumerate() {
-                match &offset_indexes[row_group_idx][column_idx] {
-                    Some(offset_index) => {
-                        let start_offset = self.buf.bytes_written();
-                        let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
-                        offset_index.write_to_out_protocol(&mut protocol)?;
-                        let end_offset = self.buf.bytes_written();
-                        // set offset and index for offset index
-                        column_metadata.offset_index_offset = Some(start_offset as i64);
-                        column_metadata.offset_index_length =
-                            Some((end_offset - start_offset) as i32);
-                    }
-                    None => {}
+                if let Some(offset_index) = &offset_indexes[row_group_idx][column_idx] {
+                    let start_offset = self.buf.bytes_written();
+                    let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
+                    offset_index.write_to_out_protocol(&mut protocol)?;
+                    let end_offset = self.buf.bytes_written();
+                    // set offset and index for offset index
+                    column_metadata.offset_index_offset = Some(start_offset as i64);
+                    column_metadata.offset_index_length = Some((end_offset - start_offset) as i32);
                 }
             }
         }
@@ -84,18 +80,14 @@ impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
         // write column index to the file
         for (row_group_idx, row_group) in self.row_groups.iter_mut().enumerate() {
             for (column_idx, column_metadata) in row_group.columns.iter_mut().enumerate() {
-                match &column_indexes[row_group_idx][column_idx] {
-                    Some(column_index) => {
-                        let start_offset = self.buf.bytes_written();
-                        let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
-                        column_index.write_to_out_protocol(&mut protocol)?;
-                        let end_offset = self.buf.bytes_written();
-                        // set offset and index for offset index
-                        column_metadata.column_index_offset = Some(start_offset as i64);
-                        column_metadata.column_index_length =
-                            Some((end_offset - start_offset) as i32);
-                    }
-                    None => {}
+                if let Some(column_index) = &column_indexes[row_group_idx][column_idx] {
+                    let start_offset = self.buf.bytes_written();
+                    let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
+                    column_index.write_to_out_protocol(&mut protocol)?;
+                    let end_offset = self.buf.bytes_written();
+                    // set offset and index for offset index
+                    column_metadata.column_index_offset = Some(start_offset as i64);
+                    column_metadata.column_index_length = Some((end_offset - start_offset) as i32);
                 }
             }
         }
@@ -524,7 +516,6 @@ mod tests {
     async fn load_metadata_from_bytes(file_size: usize, data: Bytes) -> ParquetMetaData {
         use crate::arrow::async_reader::{MetadataFetch, MetadataLoader};
         use crate::errors::Result as ParquetResult;
-        use bytes::Bytes;
         use futures::future::BoxFuture;
         use futures::FutureExt;
         use std::ops::Range;
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index 190374fd8..2665f28fe 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -294,105 +294,102 @@ impl<'a> PrimitiveTypeBuilder<'a> {
             ));
         }
 
-        match &self.logical_type {
-            Some(logical_type) => {
-                // If a converted type is populated, check that it is consistent with
-                // its logical type
-                if self.converted_type != ConvertedType::NONE {
-                    if ConvertedType::from(self.logical_type.clone()) != self.converted_type {
-                        return Err(general_err!(
-                            "Logical type {:?} is incompatible with converted type {} for field '{}'",
-                            logical_type,
-                            self.converted_type,
-                            self.name
-                        ));
-                    }
-                } else {
-                    // Populate the converted type for backwards compatibility
-                    basic_info.converted_type = self.logical_type.clone().into();
+        if let Some(logical_type) = &self.logical_type {
+            // If a converted type is populated, check that it is consistent with
+            // its logical type
+            if self.converted_type != ConvertedType::NONE {
+                if ConvertedType::from(self.logical_type.clone()) != self.converted_type {
+                    return Err(general_err!(
+                        "Logical type {:?} is incompatible with converted type {} for field '{}'",
+                        logical_type,
+                        self.converted_type,
+                        self.name
+                    ));
+                }
+            } else {
+                // Populate the converted type for backwards compatibility
+                basic_info.converted_type = self.logical_type.clone().into();
+            }
+            // Check that logical type and physical type are compatible
+            match (logical_type, self.physical_type) {
+                (LogicalType::Map, _) | (LogicalType::List, _) => {
+                    return Err(general_err!(
+                        "{:?} cannot be applied to a primitive type for field '{}'",
+                        logical_type,
+                        self.name
+                    ));
                 }
-                // Check that logical type and physical type are compatible
-                match (logical_type, self.physical_type) {
-                    (LogicalType::Map, _) | (LogicalType::List, _) => {
+                (LogicalType::Enum, PhysicalType::BYTE_ARRAY) => {}
+                (LogicalType::Decimal { scale, precision }, _) => {
+                    // Check that scale and precision are consistent with legacy values
+                    if *scale != self.scale {
                         return Err(general_err!(
-                            "{:?} cannot be applied to a primitive type for field '{}'",
-                            logical_type,
+                            "DECIMAL logical type scale {} must match self.scale {} for field '{}'",
+                            scale,
+                            self.scale,
                             self.name
                         ));
                     }
-                    (LogicalType::Enum, PhysicalType::BYTE_ARRAY) => {}
-                    (LogicalType::Decimal { scale, precision }, _) => {
-                        // Check that scale and precision are consistent with legacy values
-                        if *scale != self.scale {
-                            return Err(general_err!(
-                                "DECIMAL logical type scale {} must match self.scale {} for field '{}'",
-                                scale,
-                                self.scale,
-                                self.name
-                            ));
-                        }
-                        if *precision != self.precision {
-                            return Err(general_err!(
-                                "DECIMAL logical type precision {} must match self.precision {} for field '{}'",
-                                precision,
-                                self.precision,
-                                self.name
-                            ));
-                        }
-                        self.check_decimal_precision_scale()?;
-                    }
-                    (LogicalType::Date, PhysicalType::INT32) => {}
-                    (
-                        LogicalType::Time {
-                            unit: TimeUnit::MILLIS(_),
-                            ..
-                        },
-                        PhysicalType::INT32,
-                    ) => {}
-                    (LogicalType::Time { unit, .. }, PhysicalType::INT64) => {
-                        if *unit == TimeUnit::MILLIS(Default::default()) {
-                            return Err(general_err!(
-                                "Cannot use millisecond unit on INT64 type for field '{}'",
-                                self.name
-                            ));
-                        }
-                    }
-                    (LogicalType::Timestamp { .. }, PhysicalType::INT64) => {}
-                    (LogicalType::Integer { bit_width, .. }, PhysicalType::INT32)
-                        if *bit_width <= 32 => {}
-                    (LogicalType::Integer { bit_width, .. }, PhysicalType::INT64)
-                        if *bit_width == 64 => {}
-                    // Null type
-                    (LogicalType::Unknown, PhysicalType::INT32) => {}
-                    (LogicalType::String, PhysicalType::BYTE_ARRAY) => {}
-                    (LogicalType::Json, PhysicalType::BYTE_ARRAY) => {}
-                    (LogicalType::Bson, PhysicalType::BYTE_ARRAY) => {}
-                    (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) if self.length == 16 => {}
-                    (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) => {
-                        return Err(general_err!(
-                            "UUID cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(16) field",
-                            self.name
-                        ))
-                    }
-                    (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY)
-                        if self.length == 2 => {}
-                    (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY) => {
+                    if *precision != self.precision {
                         return Err(general_err!(
-                            "FLOAT16 cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(2) field",
+                            "DECIMAL logical type precision {} must match self.precision {} for field '{}'",
+                            precision,
+                            self.precision,
                             self.name
-                        ))
+                        ));
                     }
-                    (a, b) => {
+                    self.check_decimal_precision_scale()?;
+                }
+                (LogicalType::Date, PhysicalType::INT32) => {}
+                (
+                    LogicalType::Time {
+                        unit: TimeUnit::MILLIS(_),
+                        ..
+                    },
+                    PhysicalType::INT32,
+                ) => {}
+                (LogicalType::Time { unit, .. }, PhysicalType::INT64) => {
+                    if *unit == TimeUnit::MILLIS(Default::default()) {
                         return Err(general_err!(
-                            "Cannot annotate {:?} from {} for field '{}'",
-                            a,
-                            b,
+                            "Cannot use millisecond unit on INT64 type for field '{}'",
                             self.name
-                        ))
+                        ));
                     }
                 }
+                (LogicalType::Timestamp { .. }, PhysicalType::INT64) => {}
+                (LogicalType::Integer { bit_width, .. }, PhysicalType::INT32)
+                    if *bit_width <= 32 => {}
+                (LogicalType::Integer { bit_width, .. }, PhysicalType::INT64)
+                    if *bit_width == 64 => {}
+                // Null type
+                (LogicalType::Unknown, PhysicalType::INT32) => {}
+                (LogicalType::String, PhysicalType::BYTE_ARRAY) => {}
+                (LogicalType::Json, PhysicalType::BYTE_ARRAY) => {}
+                (LogicalType::Bson, PhysicalType::BYTE_ARRAY) => {}
+                (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) if self.length == 16 => {}
+                (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) => {
+                    return Err(general_err!(
+                        "UUID cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(16) field",
+                        self.name
+                    ))
+                }
+                (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY)
+                    if self.length == 2 => {}
+                (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY) => {
+                    return Err(general_err!(
+                        "FLOAT16 cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(2) field",
+                        self.name
+                    ))
+                }
+                (a, b) => {
+                    return Err(general_err!(
+                        "Cannot annotate {:?} from {} for field '{}'",
+                        a,
+                        b,
+                        self.name
+                    ))
+                }
             }
-            None => {}
         }
 
         match self.converted_type {
diff --git a/parquet/src/util/bit_util.rs b/parquet/src/util/bit_util.rs
index a17202254..062f93270 100644
--- a/parquet/src/util/bit_util.rs
+++ b/parquet/src/util/bit_util.rs
@@ -1027,7 +1027,10 @@ mod tests {
             .collect();
 
         // Generic values used to check against actual values read from `get_batch`.
-        let expected_values: Vec<T> = values.iter().map(|v| T::try_from_le_slice(v.as_bytes()).unwrap()).collect();
+        let expected_values: Vec<T> = values
+            .iter()
+            .map(|v| T::try_from_le_slice(v.as_bytes()).unwrap())
+            .collect();
 
         (0..total).for_each(|i| writer.put_value(values[i], num_bits));
 
diff --git a/parquet/src/util/interner.rs b/parquet/src/util/interner.rs
index f57fc3a71..a804419b5 100644
--- a/parquet/src/util/interner.rs
+++ b/parquet/src/util/interner.rs
@@ -35,7 +35,7 @@ pub trait Storage {
 
     /// Return an estimate of the memory used in this storage, in bytes
     #[allow(dead_code)] // not used in parquet_derive, so is dead there
-   fn estimated_memory_size(&self) -> usize;
+    fn estimated_memory_size(&self) -> usize;
 }
 
 /// A generic value interner supporting various different [`Storage`]


Test patch:
diff --git a/arrow-flight/tests/flight_sql_client_cli.rs b/arrow-flight/tests/flight_sql_client_cli.rs
index 168015d07..6e1f6142c 100644
--- a/arrow-flight/tests/flight_sql_client_cli.rs
+++ b/arrow-flight/tests/flight_sql_client_cli.rs
@@ -23,10 +23,12 @@ use crate::common::fixture::TestFixture;
 use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
 use arrow_flight::{
     decode::FlightRecordBatchStream,
+    encode::FlightDataEncoderBuilder,
     flight_service_server::{FlightService, FlightServiceServer},
     sql::{
         server::{FlightSqlService, PeekableFlightDataStream},
         ActionCreatePreparedStatementRequest, ActionCreatePreparedStatementResult, Any,
+        CommandGetCatalogs, CommandGetDbSchemas, CommandGetTableTypes, CommandGetTables,
         CommandPreparedStatementQuery, CommandStatementQuery, DoPutPreparedStatementResult,
         ProstMessageExt, SqlInfo,
     },
@@ -85,6 +87,205 @@ async fn test_simple() {
     );
 }
 
+#[tokio::test]
+async fn test_get_catalogs() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("catalogs")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+\
+        \n| catalog_name |\
+        \n+--------------+\
+        \n| catalog_a    |\
+        \n| catalog_b    |\
+        \n+--------------+",
+    );
+}
+
+#[tokio::test]
+async fn test_get_db_schemas() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("db-schemas")
+            .arg("catalog_a")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+----------------+\
+        \n| catalog_name | db_schema_name |\
+        \n+--------------+----------------+\
+        \n| catalog_a    | schema_1       |\
+        \n| catalog_a    | schema_2       |\
+        \n+--------------+----------------+",
+    );
+}
+
+#[tokio::test]
+async fn test_get_tables() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("tables")
+            .arg("catalog_a")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+----------------+------------+------------+\
+        \n| catalog_name | db_schema_name | table_name | table_type |\
+        \n+--------------+----------------+------------+------------+\
+        \n| catalog_a    | schema_1       | table_1    | TABLE      |\
+        \n| catalog_a    | schema_2       | table_2    | VIEW       |\
+        \n+--------------+----------------+------------+------------+",
+    );
+}
+#[tokio::test]
+async fn test_get_tables_db_filter() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("tables")
+            .arg("catalog_a")
+            .arg("--db-schema-filter")
+            .arg("schema_2")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+----------------+------------+------------+\
+        \n| catalog_name | db_schema_name | table_name | table_type |\
+        \n+--------------+----------------+------------+------------+\
+        \n| catalog_a    | schema_2       | table_2    | VIEW       |\
+        \n+--------------+----------------+------------+------------+",
+    );
+}
+
+#[tokio::test]
+async fn test_get_tables_types() {
+    let test_server = FlightSqlServiceImpl::default();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let addr = fixture.addr;
+
+    let stdout = tokio::task::spawn_blocking(move || {
+        Command::cargo_bin("flight_sql_client")
+            .unwrap()
+            .env_clear()
+            .env("RUST_BACKTRACE", "1")
+            .env("RUST_LOG", "warn")
+            .arg("--host")
+            .arg(addr.ip().to_string())
+            .arg("--port")
+            .arg(addr.port().to_string())
+            .arg("table-types")
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    })
+    .await
+    .unwrap();
+
+    fixture.shutdown_and_wait().await;
+
+    assert_eq!(
+        std::str::from_utf8(&stdout).unwrap().trim(),
+        "+--------------+\
+        \n| table_type   |\
+        \n+--------------+\
+        \n| SYSTEM_TABLE |\
+        \n| TABLE        |\
+        \n| VIEW         |\
+        \n+--------------+",
+    );
+}
+
 const PREPARED_QUERY: &str = "SELECT * FROM table WHERE field = $1";
 const PREPARED_STATEMENT_HANDLE: &str = "prepared_statement_handle";
 const UPDATED_PREPARED_STATEMENT_HANDLE: &str = "updated_prepared_statement_handle";
@@ -278,6 +479,84 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(resp)
     }
 
+    async fn get_flight_info_catalogs(
+        &self,
+        query: CommandGetCatalogs,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
+    async fn get_flight_info_schemas(
+        &self,
+        query: CommandGetDbSchemas,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
+
+    async fn get_flight_info_tables(
+        &self,
+        query: CommandGetTables,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
+
+    async fn get_flight_info_table_types(
+        &self,
+        query: CommandGetTableTypes,
+        request: Request<FlightDescriptor>,
+    ) -> Result<Response<FlightInfo>, Status> {
+        let flight_descriptor = request.into_inner();
+        let ticket = Ticket {
+            ticket: query.as_any().encode_to_vec().into(),
+        };
+        let endpoint = FlightEndpoint::new().with_ticket(ticket);
+
+        let flight_info = FlightInfo::new()
+            .try_with_schema(&query.into_builder().schema())
+            .unwrap()
+            .with_endpoint(endpoint)
+            .with_descriptor(flight_descriptor);
+
+        Ok(Response::new(flight_info))
+    }
     async fn get_flight_info_statement(
         &self,
         query: CommandStatementQuery,
@@ -309,6 +588,109 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(resp)
     }
 
+    async fn do_get_catalogs(
+        &self,
+        query: CommandGetCatalogs,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for catalog_name in ["catalog_a", "catalog_b"] {
+            builder.append(catalog_name);
+        }
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    async fn do_get_schemas(
+        &self,
+        query: CommandGetDbSchemas,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for (catalog_name, schema_name) in [
+            ("catalog_a", "schema_1"),
+            ("catalog_a", "schema_2"),
+            ("catalog_b", "schema_3"),
+        ] {
+            builder.append(catalog_name, schema_name);
+        }
+
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    async fn do_get_tables(
+        &self,
+        query: CommandGetTables,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for (catalog_name, schema_name, table_name, table_type, schema) in [
+            (
+                "catalog_a",
+                "schema_1",
+                "table_1",
+                "TABLE",
+                Arc::new(Schema::empty()),
+            ),
+            (
+                "catalog_a",
+                "schema_2",
+                "table_2",
+                "VIEW",
+                Arc::new(Schema::empty()),
+            ),
+            (
+                "catalog_b",
+                "schema_3",
+                "table_3",
+                "TABLE",
+                Arc::new(Schema::empty()),
+            ),
+        ] {
+            builder
+                .append(catalog_name, schema_name, table_name, table_type, &schema)
+                .unwrap();
+        }
+
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    async fn do_get_table_types(
+        &self,
+        query: CommandGetTableTypes,
+        _request: Request<Ticket>,
+    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
+        let mut builder = query.into_builder();
+        for table_type in ["TABLE", "VIEW", "SYSTEM_TABLE"] {
+            builder.append(table_type);
+        }
+
+        let schema = builder.schema();
+        let batch = builder.build();
+        let stream = FlightDataEncoderBuilder::new()
+            .with_schema(schema)
+            .build(futures::stream::once(async { batch }))
+            .map_err(Status::from);
+        Ok(Response::new(Box::pin(stream)))
+    }
+
     async fn do_put_prepared_statement_query(
         &self,
         _query: CommandPreparedStatementQuery,
diff --git a/conbench/requirements-test.txt b/conbench/requirements-test.txt
deleted file mode 100644
index 5e5647acd..000000000
--- a/conbench/requirements-test.txt
+++ /dev/null
@@ -1,3 +0,0 @@
-black
-flake8
-isort
diff --git a/parquet/src/util/test_common/file_util.rs b/parquet/src/util/test_common/file_util.rs
index c2dcd6773..6c031358e 100644
--- a/parquet/src/util/test_common/file_util.rs
+++ b/parquet/src/util/test_common/file_util.rs
@@ -19,8 +19,7 @@ use std::{fs, path::PathBuf, str::FromStr};
 
 /// Returns path to the test parquet file in 'data' directory
 pub fn get_test_path(file_name: &str) -> PathBuf {
-    let mut pathbuf =
-        PathBuf::from_str(&arrow::util::test_util::parquet_test_data()).unwrap();
+    let mut pathbuf = PathBuf::from_str(&arrow::util::test_util::parquet_test_data()).unwrap();
     pathbuf.push(file_name);
     pathbuf
 }
diff --git a/parquet/src/util/test_common/mod.rs b/parquet/src/util/test_common/mod.rs
index 504219eca..8cfc1e6dd 100644
--- a/parquet/src/util/test_common/mod.rs
+++ b/parquet/src/util/test_common/mod.rs
@@ -21,4 +21,4 @@ pub mod page_util;
 pub mod file_util;
 
 #[cfg(test)]
-pub mod rand_gen;
\ No newline at end of file
+pub mod rand_gen;
diff --git a/parquet/src/util/test_common/rand_gen.rs b/parquet/src/util/test_common/rand_gen.rs
index a267c3484..ec80d3a59 100644
--- a/parquet/src/util/test_common/rand_gen.rs
+++ b/parquet/src/util/test_common/rand_gen.rs
@@ -173,8 +173,7 @@ pub fn make_pages<T: DataType>(
 
         // Generate the current page
 
-        let mut pb =
-            DataPageBuilderImpl::new(desc.clone(), num_values_cur_page as u32, use_v2);
+        let mut pb = DataPageBuilderImpl::new(desc.clone(), num_values_cur_page as u32, use_v2);
         if max_rep_level > 0 {
             pb.add_rep_levels(max_rep_level, &rep_levels[level_range.clone()]);
         }