Repository: apache/arrow-rs
Pull request #6265 (2024-08-16 20:07:16)

object_store/delimited: Fix `TrailingEscape` condition
This seems like a copy-paste mistake since checking `is_quote` twice is probably wrong...




Patch:
diff --git a/CHANGELOG-old.md b/CHANGELOG-old.md
index 8b70fea39..4808cde70 100644
--- a/CHANGELOG-old.md
+++ b/CHANGELOG-old.md
@@ -19,6 +19,73 @@
 
 # Historical Changelog
 
+## [52.2.0](https://github.com/apache/arrow-rs/tree/52.2.0) (2024-07-24)
+
+[Full Changelog](https://github.com/apache/arrow-rs/compare/52.1.0...52.2.0)
+
+**Implemented enhancements:**
+
+- Faster min/max for string/binary view arrays [\#6088](https://github.com/apache/arrow-rs/issues/6088) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Support casting to/from Utf8View [\#6076](https://github.com/apache/arrow-rs/issues/6076) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Min/max support for String/BinaryViewArray [\#6052](https://github.com/apache/arrow-rs/issues/6052) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Improve performance of constructing `ByteView`s for small strings [\#6034](https://github.com/apache/arrow-rs/issues/6034) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Fast UTF-8 validation when reading StringViewArray from Parquet [\#5995](https://github.com/apache/arrow-rs/issues/5995) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Optimize StringView row decoding [\#5945](https://github.com/apache/arrow-rs/issues/5945) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Implementing `deduplicate` / `intern` functionality for StringView [\#5910](https://github.com/apache/arrow-rs/issues/5910) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Add `FlightSqlServiceClient::new_from_inner` [\#6003](https://github.com/apache/arrow-rs/pull/6003) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
+- Complete `StringViewArray` and `BinaryViewArray` parquet decoder:  [\#6004](https://github.com/apache/arrow-rs/pull/6004) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Add begin/end\_transaction methods in FlightSqlServiceClient [\#6026](https://github.com/apache/arrow-rs/pull/6026) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
+- Read Parquet statistics as arrow `Arrays`  [\#6046](https://github.com/apache/arrow-rs/pull/6046) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([efredine](https://github.com/efredine))
+
+**Fixed bugs:**
+
+- Panic in `ParquetMetadata::memory_size` if no min/max set [\#6091](https://github.com/apache/arrow-rs/issues/6091) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- BinaryViewArray doesn't roundtrip a single `Some(&[])` through parquet [\#6086](https://github.com/apache/arrow-rs/issues/6086) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Parquet `ColumnIndex` for null columns is written even when statistics are disabled [\#6010](https://github.com/apache/arrow-rs/issues/6010) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+
+**Documentation updates:**
+
+- Fix typo in GenericByteViewArray documentation [\#6054](https://github.com/apache/arrow-rs/pull/6054) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([progval](https://github.com/progval))
+- Minor: Improve parquet PageIndex documentation [\#6042](https://github.com/apache/arrow-rs/pull/6042) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+
+**Closed issues:**
+
+- Potential performance improvements for reading Parquet to StringViewArray/BinaryViewArray [\#5904](https://github.com/apache/arrow-rs/issues/5904) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+
+**Merged pull requests:**
+
+- Faster `GenericByteView` construction [\#6102](https://github.com/apache/arrow-rs/pull/6102) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Add benchmark to track byte-view construction performance [\#6101](https://github.com/apache/arrow-rs/pull/6101) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Optimize `bool_or` using `max_boolean` [\#6100](https://github.com/apache/arrow-rs/pull/6100) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([simonvandel](https://github.com/simonvandel))
+- Optimize `max_boolean` by operating on u64 chunks [\#6098](https://github.com/apache/arrow-rs/pull/6098) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([simonvandel](https://github.com/simonvandel))
+- fix panic in `ParquetMetadata::memory_size`: check has\_min\_max\_set before invoking min\(\)/max\(\) [\#6092](https://github.com/apache/arrow-rs/pull/6092) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([Fischer0522](https://github.com/Fischer0522))
+- Implement specialized min/max for `GenericBinaryView` \(`StringView` and `BinaryView`\) [\#6089](https://github.com/apache/arrow-rs/pull/6089) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Add PartialEq to ParquetMetaData and FileMetadata [\#6082](https://github.com/apache/arrow-rs/pull/6082) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([adriangb](https://github.com/adriangb))
+- Enable casting from Utf8View [\#6077](https://github.com/apache/arrow-rs/pull/6077) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([a10y](https://github.com/a10y))
+- StringView support in arrow-csv [\#6062](https://github.com/apache/arrow-rs/pull/6062) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([2010YOUY01](https://github.com/2010YOUY01))
+- Implement min max support for string/binary view types [\#6053](https://github.com/apache/arrow-rs/pull/6053) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Minor: clarify the relationship between `file::metadata` and `format` in docs [\#6049](https://github.com/apache/arrow-rs/pull/6049) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Minor API adjustments for StringViewBuilder [\#6047](https://github.com/apache/arrow-rs/pull/6047) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Add parquet `StatisticsConverter` for arrow reader [\#6046](https://github.com/apache/arrow-rs/pull/6046) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([efredine](https://github.com/efredine))
+- Directly decode String/BinaryView types from arrow-row format [\#6044](https://github.com/apache/arrow-rs/pull/6044) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Clean up unused code for view types in offset buffer [\#6040](https://github.com/apache/arrow-rs/pull/6040) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Avoid using Buffer api that accidentally copies data [\#6039](https://github.com/apache/arrow-rs/pull/6039) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([XiangpengHao](https://github.com/XiangpengHao))
+- MINOR: Fix `hashbrown` version in `arrow-array`, remove from `arrow-row` [\#6035](https://github.com/apache/arrow-rs/pull/6035) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([mbrobbel](https://github.com/mbrobbel))
+- Improve performance reading `ByteViewArray` from parquet by removing an implicit copy [\#6031](https://github.com/apache/arrow-rs/pull/6031) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Add begin/end\_transaction methods in FlightSqlServiceClient [\#6026](https://github.com/apache/arrow-rs/pull/6026) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
+- Unsafe improvements: core `parquet` crate. [\#6024](https://github.com/apache/arrow-rs/pull/6024) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([veluca93](https://github.com/veluca93))
+- Additional tests for parquet reader utf8 validation [\#6023](https://github.com/apache/arrow-rs/pull/6023) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Update zstd-sys requirement from \>=2.0.0, \<2.0.12 to \>=2.0.0, \<2.0.13 [\#6019](https://github.com/apache/arrow-rs/pull/6019) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- fix doc ci in latest rust nightly version [\#6012](https://github.com/apache/arrow-rs/pull/6012) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([Rachelint](https://github.com/Rachelint))
+- Do not write `ColumnIndex` for null columns when not writing page statistics [\#6011](https://github.com/apache/arrow-rs/pull/6011) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Fast utf8 validation when loading string view from parquet [\#6009](https://github.com/apache/arrow-rs/pull/6009) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Deduplicate strings/binarys when building view types [\#6005](https://github.com/apache/arrow-rs/pull/6005) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Complete `StringViewArray` and `BinaryViewArray` parquet decoder:  implement delta byte array and delta length byte array encoding [\#6004](https://github.com/apache/arrow-rs/pull/6004) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Add `FlightSqlServiceClient::new_from_inner` [\#6003](https://github.com/apache/arrow-rs/pull/6003) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
+- Rename `Schema::all_fields` to `flattened_fields` [\#6001](https://github.com/apache/arrow-rs/pull/6001) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
+- Refine documentation and examples for `DataType` [\#5997](https://github.com/apache/arrow-rs/pull/5997) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- implement `DataType::try_form(&str)` [\#5994](https://github.com/apache/arrow-rs/pull/5994) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
+- Implement dictionary support for reading ByteView from parquet [\#5973](https://github.com/apache/arrow-rs/pull/5973) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
 ## [52.1.0](https://github.com/apache/arrow-rs/tree/52.1.0) (2024-07-02)
 
 [Full Changelog](https://github.com/apache/arrow-rs/compare/52.0.0...52.1.0)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a1ac7deef..72b809019 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,73 +19,160 @@
 
 # Changelog
 
-## [52.2.0](https://github.com/apache/arrow-rs/tree/52.2.0) (2024-07-24)
+## [53.0.0](https://github.com/apache/arrow-rs/tree/53.0.0) (2024-08-31)
 
-[Full Changelog](https://github.com/apache/arrow-rs/compare/52.1.0...52.2.0)
+[Full Changelog](https://github.com/apache/arrow-rs/compare/52.2.0...53.0.0)
+
+**Breaking changes:**
+
+- parquet\_derive: Match fields by name, support reading selected fields rather than all [\#6269](https://github.com/apache/arrow-rs/pull/6269) ([double-free](https://github.com/double-free))
+- Update parquet object\_store dependency to 0.11.0 [\#6264](https://github.com/apache/arrow-rs/pull/6264) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- parquet Statistics - deprecate `has_*` APIs and add `_opt` functions that return `Option<T>` [\#6216](https://github.com/apache/arrow-rs/pull/6216) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([Michael-J-Ward](https://github.com/Michael-J-Ward))
+- Expose bulk ingest in flight sql client and server [\#6201](https://github.com/apache/arrow-rs/pull/6201) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([djanderson](https://github.com/djanderson))
+- 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))
 
 **Implemented enhancements:**
 
-- Faster min/max for string/binary view arrays [\#6088](https://github.com/apache/arrow-rs/issues/6088) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
-- Support casting to/from Utf8View [\#6076](https://github.com/apache/arrow-rs/issues/6076) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
-- Min/max support for String/BinaryViewArray [\#6052](https://github.com/apache/arrow-rs/issues/6052) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
-- Improve performance of constructing `ByteView`s for small strings [\#6034](https://github.com/apache/arrow-rs/issues/6034) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
-- Fast UTF-8 validation when reading StringViewArray from Parquet [\#5995](https://github.com/apache/arrow-rs/issues/5995) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
-- Optimize StringView row decoding [\#5945](https://github.com/apache/arrow-rs/issues/5945) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
-- Implementing `deduplicate` / `intern` functionality for StringView [\#5910](https://github.com/apache/arrow-rs/issues/5910) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
-- Add `FlightSqlServiceClient::new_from_inner` [\#6003](https://github.com/apache/arrow-rs/pull/6003) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
-- Complete `StringViewArray` and `BinaryViewArray` parquet decoder:  [\#6004](https://github.com/apache/arrow-rs/pull/6004) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Add begin/end\_transaction methods in FlightSqlServiceClient [\#6026](https://github.com/apache/arrow-rs/pull/6026) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
-- Read Parquet statistics as arrow `Arrays`  [\#6046](https://github.com/apache/arrow-rs/pull/6046) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([efredine](https://github.com/efredine))
+- Derive `PartialEq` and `Eq` for `parquet::arrow::ProjectionMask` [\#6329](https://github.com/apache/arrow-rs/issues/6329) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Allow converting empty `pyarrow.RecordBatch` to `arrow::RecordBatch` [\#6318](https://github.com/apache/arrow-rs/issues/6318) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Parquet writer should not write any min/max data to ColumnIndex when all values are null [\#6315](https://github.com/apache/arrow-rs/issues/6315) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Parquet: Add `union` method to `RowSelection` [\#6307](https://github.com/apache/arrow-rs/issues/6307) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Support writing `UTC adjusted time` arrow array to parquet [\#6277](https://github.com/apache/arrow-rs/issues/6277) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- A better way to resize the buffer for the snappy encode/decode [\#6276](https://github.com/apache/arrow-rs/issues/6276) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- parquet\_derive: support reading selected columns from parquet file [\#6268](https://github.com/apache/arrow-rs/issues/6268)
+- Tests for invalid parquet files [\#6261](https://github.com/apache/arrow-rs/issues/6261) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Implement `date_part` for `Duration` [\#6245](https://github.com/apache/arrow-rs/issues/6245) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Avoid unnecessary null buffer construction when converting arrays to a different type [\#6243](https://github.com/apache/arrow-rs/issues/6243) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Add `parquet_opendal` in related projects [\#6235](https://github.com/apache/arrow-rs/issues/6235)
+- Look into optimizing reading FixedSizeBinary arrays from parquet [\#6219](https://github.com/apache/arrow-rs/issues/6219) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Add benchmarks for `BYTE_STREAM_SPLIT` encoded Parquet `FIXED_LEN_BYTE_ARRAY` data [\#6203](https://github.com/apache/arrow-rs/issues/6203) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Make it easy to write parquet to object\_store -- Implement `AsyncFileWriter` for a type that implements `obj_store::MultipartUpload` for `AsyncArrowWriter` [\#6200](https://github.com/apache/arrow-rs/issues/6200) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Remove test duplication in parquet statistics tets [\#6185](https://github.com/apache/arrow-rs/issues/6185) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Support BinaryView Types in C Schema FFI [\#6170](https://github.com/apache/arrow-rs/issues/6170) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- speedup take\_byte\_view kernel [\#6167](https://github.com/apache/arrow-rs/issues/6167) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Add support for `StringView` and `BinaryView` statistics in `StatisticsConverter` [\#6164](https://github.com/apache/arrow-rs/issues/6164) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Support casting `BinaryView` --\> `Utf8` and `LargeUtf8` [\#6162](https://github.com/apache/arrow-rs/issues/6162) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Implement `filter` kernel specially for `FixedSizeByteArray` [\#6153](https://github.com/apache/arrow-rs/issues/6153) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Use `LevelHistogram` throughout Parquet metadata [\#6134](https://github.com/apache/arrow-rs/issues/6134) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Support DoPutStatementIngest from Arrow Flight SQL 17.0 [\#6124](https://github.com/apache/arrow-rs/issues/6124) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)]
+- ColumnMetaData should no longer be written inline with data [\#6115](https://github.com/apache/arrow-rs/issues/6115) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Implement date\_part for `Interval`  [\#6113](https://github.com/apache/arrow-rs/issues/6113) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Implement `Into<Arc<dyn Array>>` for `ArrayData` [\#6104](https://github.com/apache/arrow-rs/issues/6104)
+- Allow flushing or non-buffered writes from `arrow::ipc::writer::StreamWriter` [\#6099](https://github.com/apache/arrow-rs/issues/6099) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Default block\_size for `StringViewArray` [\#6094](https://github.com/apache/arrow-rs/issues/6094) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Remove `Statistics::has_min_max_set` and `ValueStatistics::has_min_max_set` and use `Option` instead [\#6093](https://github.com/apache/arrow-rs/issues/6093) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Upgrade arrow-flight to tonic 0.12 [\#6072](https://github.com/apache/arrow-rs/issues/6072)
+- Improve speed of row converter by skipping utf8 checks [\#6058](https://github.com/apache/arrow-rs/issues/6058) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Extend support for BYTE\_STREAM\_SPLIT to FIXED\_LEN\_BYTE\_ARRAY, INT32, and INT64 primitive types [\#6048](https://github.com/apache/arrow-rs/issues/6048) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Release arrow-rs / parquet minor version `52.2.0` \(August 2024\) [\#5998](https://github.com/apache/arrow-rs/issues/5998) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
 
 **Fixed bugs:**
 
-- Panic in `ParquetMetadata::memory_size` if no min/max set [\#6091](https://github.com/apache/arrow-rs/issues/6091) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
-- BinaryViewArray doesn't roundtrip a single `Some(&[])` through parquet [\#6086](https://github.com/apache/arrow-rs/issues/6086) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
-- Parquet `ColumnIndex` for null columns is written even when statistics are disabled [\#6010](https://github.com/apache/arrow-rs/issues/6010) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Invalid `ColumnIndex` written in parquet [\#6310](https://github.com/apache/arrow-rs/issues/6310) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- comparison\_kernels benchmarks panic [\#6283](https://github.com/apache/arrow-rs/issues/6283) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Printing schema metadata includes possibly incorrect compression level [\#6270](https://github.com/apache/arrow-rs/issues/6270) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Don't panic when creating `Field` from `FFI_ArrowSchema` with no name [\#6251](https://github.com/apache/arrow-rs/issues/6251) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- lexsort\_to\_indices should not fallback to non-lexical sort if the datatype is not supported [\#6226](https://github.com/apache/arrow-rs/issues/6226) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Parquet Statistics null\_count does not distinguish between `0` and not specified [\#6215](https://github.com/apache/arrow-rs/issues/6215) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
+- Using a take kernel on a dense union can result in reaching "unreachable" code [\#6206](https://github.com/apache/arrow-rs/issues/6206) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Adding sub day seconds to Date64 is ignored. [\#6198](https://github.com/apache/arrow-rs/issues/6198) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- mismatch between parquet type `is_optional` codes and comment [\#6191](https://github.com/apache/arrow-rs/issues/6191) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)]
 
 **Documentation updates:**
 
-- Fix typo in GenericByteViewArray documentation [\#6054](https://github.com/apache/arrow-rs/pull/6054) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([progval](https://github.com/progval))
-- Minor: Improve parquet PageIndex documentation [\#6042](https://github.com/apache/arrow-rs/pull/6042) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Minor: improve filter documentation [\#6317](https://github.com/apache/arrow-rs/pull/6317) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Minor: Improve comments on GenericByteViewArray::bytes\_iter\(\), prefix\_iter\(\) and suffix\_iter\(\) [\#6306](https://github.com/apache/arrow-rs/pull/6306) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Minor: improve `RowFilter` and `ArrowPredicate` docs [\#6301](https://github.com/apache/arrow-rs/pull/6301) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Improve documentation for `MutableArrayData` [\#6272](https://github.com/apache/arrow-rs/pull/6272) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Add examples to `StringViewBuilder` and `BinaryViewBuilder` [\#6240](https://github.com/apache/arrow-rs/pull/6240) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- minor: enhance document for ParquetField [\#6239](https://github.com/apache/arrow-rs/pull/6239) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([mapleFU](https://github.com/mapleFU))
+- Minor: Improve Type documentation [\#6224](https://github.com/apache/arrow-rs/pull/6224) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Minor: Update `DateType::Date64` docs [\#6223](https://github.com/apache/arrow-rs/pull/6223) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Add \(more\) Parquet Metadata Documentation [\#6184](https://github.com/apache/arrow-rs/pull/6184) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Add additional documentation and examples to `ArrayAccessor` [\#6141](https://github.com/apache/arrow-rs/pull/6141) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Minor: improve comments in temporal.rs tests [\#6140](https://github.com/apache/arrow-rs/pull/6140) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Minor: Update release schedule in README [\#6125](https://github.com/apache/arrow-rs/pull/6125) ([alamb](https://github.com/alamb))
 
 **Closed issues:**
 
-- Potential performance improvements for reading Parquet to StringViewArray/BinaryViewArray [\#5904](https://github.com/apache/arrow-rs/issues/5904) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
+- Simplify take octokit workflow [\#6279](https://github.com/apache/arrow-rs/issues/6279)
+- Make the bearer token visible in FlightSqlServiceClient [\#6253](https://github.com/apache/arrow-rs/issues/6253) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)]
+- Port `take` workflow to use `oktokit` [\#6242](https://github.com/apache/arrow-rs/issues/6242)
+- Remove `SchemaBuilder` dependency from `StructArray` constructors [\#6138](https://github.com/apache/arrow-rs/issues/6138) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)]
 
 **Merged pull requests:**
 
-- Faster `GenericByteView` construction [\#6102](https://github.com/apache/arrow-rs/pull/6102) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Add benchmark to track byte-view construction performance [\#6101](https://github.com/apache/arrow-rs/pull/6101) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Optimize `bool_or` using `max_boolean` [\#6100](https://github.com/apache/arrow-rs/pull/6100) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([simonvandel](https://github.com/simonvandel))
-- Optimize `max_boolean` by operating on u64 chunks [\#6098](https://github.com/apache/arrow-rs/pull/6098) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([simonvandel](https://github.com/simonvandel))
-- fix panic in `ParquetMetadata::memory_size`: check has\_min\_max\_set before invoking min\(\)/max\(\) [\#6092](https://github.com/apache/arrow-rs/pull/6092) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([Fischer0522](https://github.com/Fischer0522))
-- Implement specialized min/max for `GenericBinaryView` \(`StringView` and `BinaryView`\) [\#6089](https://github.com/apache/arrow-rs/pull/6089) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Add PartialEq to ParquetMetaData and FileMetadata [\#6082](https://github.com/apache/arrow-rs/pull/6082) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([adriangb](https://github.com/adriangb))
-- Enable casting from Utf8View [\#6077](https://github.com/apache/arrow-rs/pull/6077) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([a10y](https://github.com/a10y))
-- StringView support in arrow-csv [\#6062](https://github.com/apache/arrow-rs/pull/6062) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([2010YOUY01](https://github.com/2010YOUY01))
-- Implement min max support for string/binary view types [\#6053](https://github.com/apache/arrow-rs/pull/6053) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Minor: clarify the relationship between `file::metadata` and `format` in docs [\#6049](https://github.com/apache/arrow-rs/pull/6049) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
-- Minor API adjustments for StringViewBuilder [\#6047](https://github.com/apache/arrow-rs/pull/6047) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Add parquet `StatisticsConverter` for arrow reader [\#6046](https://github.com/apache/arrow-rs/pull/6046) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([efredine](https://github.com/efredine))
-- Directly decode String/BinaryView types from arrow-row format [\#6044](https://github.com/apache/arrow-rs/pull/6044) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Clean up unused code for view types in offset buffer [\#6040](https://github.com/apache/arrow-rs/pull/6040) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Avoid using Buffer api that accidentally copies data [\#6039](https://github.com/apache/arrow-rs/pull/6039) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([XiangpengHao](https://github.com/XiangpengHao))
-- MINOR: Fix `hashbrown` version in `arrow-array`, remove from `arrow-row` [\#6035](https://github.com/apache/arrow-rs/pull/6035) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([mbrobbel](https://github.com/mbrobbel))
-- Improve performance reading `ByteViewArray` from parquet by removing an implicit copy [\#6031](https://github.com/apache/arrow-rs/pull/6031) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Add begin/end\_transaction methods in FlightSqlServiceClient [\#6026](https://github.com/apache/arrow-rs/pull/6026) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
-- Unsafe improvements: core `parquet` crate. [\#6024](https://github.com/apache/arrow-rs/pull/6024) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([veluca93](https://github.com/veluca93))
-- Additional tests for parquet reader utf8 validation [\#6023](https://github.com/apache/arrow-rs/pull/6023) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
-- Update zstd-sys requirement from \>=2.0.0, \<2.0.12 to \>=2.0.0, \<2.0.13 [\#6019](https://github.com/apache/arrow-rs/pull/6019) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([dependabot[bot]](https://github.com/apps/dependabot))
-- fix doc ci in latest rust nightly version [\#6012](https://github.com/apache/arrow-rs/pull/6012) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([Rachelint](https://github.com/Rachelint))
-- Do not write `ColumnIndex` for null columns when not writing page statistics [\#6011](https://github.com/apache/arrow-rs/pull/6011) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
-- Fast utf8 validation when loading string view from parquet [\#6009](https://github.com/apache/arrow-rs/pull/6009) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Deduplicate strings/binarys when building view types [\#6005](https://github.com/apache/arrow-rs/pull/6005) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Complete `StringViewArray` and `BinaryViewArray` parquet decoder:  implement delta byte array and delta length byte array encoding [\#6004](https://github.com/apache/arrow-rs/pull/6004) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
-- Add `FlightSqlServiceClient::new_from_inner` [\#6003](https://github.com/apache/arrow-rs/pull/6003) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
-- Rename `Schema::all_fields` to `flattened_fields` [\#6001](https://github.com/apache/arrow-rs/pull/6001) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([lewiszlw](https://github.com/lewiszlw))
-- Refine documentation and examples for `DataType` [\#5997](https://github.com/apache/arrow-rs/pull/5997) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
-- implement `DataType::try_form(&str)` [\#5994](https://github.com/apache/arrow-rs/pull/5994) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
-- Implement dictionary support for reading ByteView from parquet [\#5973](https://github.com/apache/arrow-rs/pull/5973) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Derive PartialEq and Eq for parquet::arrow::ProjectionMask [\#6330](https://github.com/apache/arrow-rs/pull/6330) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([thinkharderdev](https://github.com/thinkharderdev))
+- Support zero column `RecordBatch`es in pyarrow integration \(use RecordBatchOptions when converting a pyarrow RecordBatch\) [\#6320](https://github.com/apache/arrow-rs/pull/6320) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([Michael-J-Ward](https://github.com/Michael-J-Ward))
+- Fix writing of invalid  Parquet ColumnIndex when row group contains null pages [\#6319](https://github.com/apache/arrow-rs/pull/6319) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([adriangb](https://github.com/adriangb))
+- Pass empty vectors as min/max for all null pages when building ColumnIndex [\#6316](https://github.com/apache/arrow-rs/pull/6316) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Update tonic-build requirement from =0.12.0 to =0.12.2 [\#6314](https://github.com/apache/arrow-rs/pull/6314) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- Parquet: add `union` method to `RowSelection` [\#6308](https://github.com/apache/arrow-rs/pull/6308) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([sdd](https://github.com/sdd))
+- Specialize filter for structs and sparse unions [\#6304](https://github.com/apache/arrow-rs/pull/6304) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([gstvg](https://github.com/gstvg))
+- Err on `try_from_le_slice` [\#6295](https://github.com/apache/arrow-rs/pull/6295) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([samuelcolvin](https://github.com/samuelcolvin))
+- fix reference in doctest to size\_of which is not imported by default [\#6286](https://github.com/apache/arrow-rs/pull/6286) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([rtyler](https://github.com/rtyler))
+- Support writing UTC adjusted time arrays to parquet [\#6278](https://github.com/apache/arrow-rs/pull/6278) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([aykut-bozkurt](https://github.com/aykut-bozkurt))
+- Minor: `pub use ByteView` in arrow and improve documentation [\#6275](https://github.com/apache/arrow-rs/pull/6275) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Fix accessing name from ffi schema [\#6273](https://github.com/apache/arrow-rs/pull/6273) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([kylebarron](https://github.com/kylebarron))
+- Do not print compression level in schema printer [\#6271](https://github.com/apache/arrow-rs/pull/6271) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([ttencate](https://github.com/ttencate))
+- ci: use octokit to add assignee [\#6267](https://github.com/apache/arrow-rs/pull/6267) ([dsgibbons](https://github.com/dsgibbons))
+- Add tests for bad parquet files [\#6262](https://github.com/apache/arrow-rs/pull/6262) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Add `Statistics::distinct_count_opt` and deprecate `Statistics::distinct_count` [\#6259](https://github.com/apache/arrow-rs/pull/6259) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Minor: move `FallibleRequestStream` and `FallibleTonicResponseStream` to a module [\#6258](https://github.com/apache/arrow-rs/pull/6258) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([alamb](https://github.com/alamb))
+- Make the bearer token visible in FlightSqlServiceClient [\#6254](https://github.com/apache/arrow-rs/pull/6254) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([ccciudatu](https://github.com/ccciudatu))
+- Use `unary()` for array conversion in Parquet array readers, speed up `Decimal128`, `Decimal256` and `Float16`  [\#6252](https://github.com/apache/arrow-rs/pull/6252) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([etseidl](https://github.com/etseidl))
+- Update tower requirement from 0.4.13 to 0.5.0 [\#6250](https://github.com/apache/arrow-rs/pull/6250) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- Implement date\_part for durations [\#6246](https://github.com/apache/arrow-rs/pull/6246) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([nrc](https://github.com/nrc))
+- Remove unnecessary null buffer construction when converting arrays to a different type [\#6244](https://github.com/apache/arrow-rs/pull/6244) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([etseidl](https://github.com/etseidl))
+- Implement PartialEq for GenericByteViewArray [\#6241](https://github.com/apache/arrow-rs/pull/6241) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([alamb](https://github.com/alamb))
+- Minor: Remove non standard footer from LICENSE.txt / reference to Apache Aurora [\#6237](https://github.com/apache/arrow-rs/pull/6237) ([alamb](https://github.com/alamb))
+- docs: Add parquet\_opendal in related projects [\#6236](https://github.com/apache/arrow-rs/pull/6236) ([Xuanwo](https://github.com/Xuanwo))
+- Avoid infinite loop in bad parquet by checking the number of rep levels  [\#6232](https://github.com/apache/arrow-rs/pull/6232) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([jp0317](https://github.com/jp0317))
+- Specialize Prefix/Suffix Match for `Like/ILike` between Array and Scalar for StringViewArray [\#6231](https://github.com/apache/arrow-rs/pull/6231) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([xinlifoobar](https://github.com/xinlifoobar))
+- fix: lexsort\_to\_indices should not fallback to non-lexical sort if the datatype is not supported [\#6225](https://github.com/apache/arrow-rs/pull/6225) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([viirya](https://github.com/viirya))
+- Modest improvement to FixedLenByteArray BYTE\_STREAM\_SPLIT arrow decoder [\#6222](https://github.com/apache/arrow-rs/pull/6222) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Improve performance of `FixedLengthBinary` decoding  [\#6220](https://github.com/apache/arrow-rs/pull/6220) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Update documentation for Parquet BYTE\_STREAM\_SPLIT encoding [\#6212](https://github.com/apache/arrow-rs/pull/6212) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Improve interval parsing [\#6211](https://github.com/apache/arrow-rs/pull/6211) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
+- minor: Suggest take on interleave docs [\#6210](https://github.com/apache/arrow-rs/pull/6210) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([gstvg](https://github.com/gstvg))
+- fix: Correctly handle take on dense union of a single selected type [\#6209](https://github.com/apache/arrow-rs/pull/6209) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([gstvg](https://github.com/gstvg))
+- Add time dictionary coercions [\#6208](https://github.com/apache/arrow-rs/pull/6208) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([adriangb](https://github.com/adriangb))
+- fix\(arrow\): restrict the range of temporal values produced via `data_gen` [\#6205](https://github.com/apache/arrow-rs/pull/6205) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([kyle-mccarthy](https://github.com/kyle-mccarthy))
+- Add benchmarks for `BYTE_STREAM_SPLIT` encoded Parquet `FIXED_LEN_BYTE_ARRAY` data  [\#6204](https://github.com/apache/arrow-rs/pull/6204) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Move `ParquetMetadataWriter` to its own module, update documentation [\#6202](https://github.com/apache/arrow-rs/pull/6202) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- Add `ThriftMetadataWriter` for writing Parquet metadata [\#6197](https://github.com/apache/arrow-rs/pull/6197) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([adriangb](https://github.com/adriangb))
+- Update zstd-sys requirement from \>=2.0.0, \<2.0.13 to \>=2.0.0, \<2.0.14 [\#6196](https://github.com/apache/arrow-rs/pull/6196) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- fix parquet type `is_optional` comments [\#6192](https://github.com/apache/arrow-rs/pull/6192) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([jp0317](https://github.com/jp0317))
+- Remove duplicated statistics tests in parquet [\#6190](https://github.com/apache/arrow-rs/pull/6190) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([Kev1n8](https://github.com/Kev1n8))
+- Benchmarks for `bool_and` [\#6189](https://github.com/apache/arrow-rs/pull/6189) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([simonvandel](https://github.com/simonvandel))
+- Fix typo in documentation of Float64Array [\#6188](https://github.com/apache/arrow-rs/pull/6188) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([mesejo](https://github.com/mesejo))
+- Make it clear that `StatisticsConverter` can not panic [\#6187](https://github.com/apache/arrow-rs/pull/6187) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([alamb](https://github.com/alamb))
+- add filter benchmark for `FixedSizeBinaryArray` [\#6186](https://github.com/apache/arrow-rs/pull/6186) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([chloro-pn](https://github.com/chloro-pn))
+- Update sysinfo requirement from 0.30.12 to 0.31.2 [\#6182](https://github.com/apache/arrow-rs/pull/6182) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- Add support for `StringView` and `BinaryView` statistics in `StatisticsConverter` [\#6181](https://github.com/apache/arrow-rs/pull/6181) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([Kev1n8](https://github.com/Kev1n8))
+- Support casting between BinaryView \<--\> Utf8 and LargeUtf8 [\#6180](https://github.com/apache/arrow-rs/pull/6180) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([xinlifoobar](https://github.com/xinlifoobar))
+- Implement specialized filter kernel for `FixedSizeByteArray` [\#6178](https://github.com/apache/arrow-rs/pull/6178) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([chloro-pn](https://github.com/chloro-pn))
+- Support `StringView` and `BinaryView` in CDataInterface  [\#6171](https://github.com/apache/arrow-rs/pull/6171) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([a10y](https://github.com/a10y))
+- Optimize `take` kernel for `BinaryViewArray` and `StringViewArray` [\#6168](https://github.com/apache/arrow-rs/pull/6168) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([a10y](https://github.com/a10y))
+- Support Parquet `BYTE_STREAM_SPLIT` for INT32, INT64, and FIXED\_LEN\_BYTE\_ARRAY primitive types [\#6159](https://github.com/apache/arrow-rs/pull/6159) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Fix comparison kernel benchmarks [\#6147](https://github.com/apache/arrow-rs/pull/6147) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
+- improve `LIKE` regex performance up to 12x [\#6145](https://github.com/apache/arrow-rs/pull/6145) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
+- Optimize `min_boolean` and `bool_and` [\#6144](https://github.com/apache/arrow-rs/pull/6144) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([simonvandel](https://github.com/simonvandel))
+- Reduce bounds check in `RowIter`, add `unsafe Rows::row_unchecked` [\#6142](https://github.com/apache/arrow-rs/pull/6142) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Minor: Simplify `StructArray` constructors [\#6139](https://github.com/apache/arrow-rs/pull/6139) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([Rafferty97](https://github.com/Rafferty97))
+- Implement exponential block size growing strategy for `StringViewBuilder` [\#6136](https://github.com/apache/arrow-rs/pull/6136) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([XiangpengHao](https://github.com/XiangpengHao))
+- Use `LevelHistogram` in `PageIndex` [\#6135](https://github.com/apache/arrow-rs/pull/6135) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Add ArrowError::ArithmeticError [\#6130](https://github.com/apache/arrow-rs/pull/6130) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([andygrove](https://github.com/andygrove))
+- Improve `LIKE` performance for "contains" style queries [\#6128](https://github.com/apache/arrow-rs/pull/6128) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
+- Add `BooleanArray::new_from_packed` and `BooleanArray::new_from_u8` [\#6127](https://github.com/apache/arrow-rs/pull/6127) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([chloro-pn](https://github.com/chloro-pn))
+- improvements to `(i)starts_with` and `(i)ends_with` performance [\#6118](https://github.com/apache/arrow-rs/pull/6118) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([samuelcolvin](https://github.com/samuelcolvin))
+- Fix Clippy for the Rust 1.80 release [\#6116](https://github.com/apache/arrow-rs/pull/6116) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] [[arrow-flight](https://github.com/apache/arrow-rs/labels/arrow-flight)] ([alamb](https://github.com/alamb))
+- added a flush method to IPC writers [\#6108](https://github.com/apache/arrow-rs/pull/6108) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([V0ldek](https://github.com/V0ldek))
+- Add support for level histograms added in PARQUET-2261 to `ParquetMetaData` [\#6105](https://github.com/apache/arrow-rs/pull/6105) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([etseidl](https://github.com/etseidl))
+- Implement date\_part for intervals [\#6071](https://github.com/apache/arrow-rs/pull/6071) [[arrow](https://github.com/apache/arrow-rs/labels/arrow)] ([nrc](https://github.com/nrc))
+- feat\(parquet\): Implement AsyncFileWriter for `object_store::buffered::BufWriter` [\#6013](https://github.com/apache/arrow-rs/pull/6013) [[parquet](https://github.com/apache/arrow-rs/labels/parquet)] ([Xuanwo](https://github.com/Xuanwo))
 
 
 
diff --git a/Cargo.toml b/Cargo.toml
index bb7038a34..3b274d583 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -62,7 +62,7 @@ exclude = [
 ]
 
 [workspace.package]
-version = "52.2.0"
+version = "53.0.0"
 homepage = "https://github.com/apache/arrow-rs"
 repository = "https://github.com/apache/arrow-rs"
 authors = ["Apache Arrow <dev@arrow.apache.org>"]
@@ -77,20 +77,20 @@ edition = "2021"
 rust-version = "1.62"
 
 [workspace.dependencies]
-arrow = { version = "52.2.0", path = "./arrow", default-features = false }
-arrow-arith = { version = "52.2.0", path = "./arrow-arith" }
-arrow-array = { version = "52.2.0", path = "./arrow-array" }
-arrow-buffer = { version = "52.2.0", path = "./arrow-buffer" }
-arrow-cast = { version = "52.2.0", path = "./arrow-cast" }
-arrow-csv = { version = "52.2.0", path = "./arrow-csv" }
-arrow-data = { version = "52.2.0", path = "./arrow-data" }
-arrow-ipc = { version = "52.2.0", path = "./arrow-ipc" }
-arrow-json = { version = "52.2.0", path = "./arrow-json" }
-arrow-ord = { version = "52.2.0", path = "./arrow-ord" }
-arrow-row = { version = "52.2.0", path = "./arrow-row" }
-arrow-schema = { version = "52.2.0", path = "./arrow-schema" }
-arrow-select = { version = "52.2.0", path = "./arrow-select" }
-arrow-string = { version = "52.2.0", path = "./arrow-string" }
-parquet = { version = "52.2.0", path = "./parquet", default-features = false }
+arrow = { version = "53.0.0", path = "./arrow", default-features = false }
+arrow-arith = { version = "53.0.0", path = "./arrow-arith" }
+arrow-array = { version = "53.0.0", path = "./arrow-array" }
+arrow-buffer = { version = "53.0.0", path = "./arrow-buffer" }
+arrow-cast = { version = "53.0.0", path = "./arrow-cast" }
+arrow-csv = { version = "53.0.0", path = "./arrow-csv" }
+arrow-data = { version = "53.0.0", path = "./arrow-data" }
+arrow-ipc = { version = "53.0.0", path = "./arrow-ipc" }
+arrow-json = { version = "53.0.0", path = "./arrow-json" }
+arrow-ord = { version = "53.0.0", path = "./arrow-ord" }
+arrow-row = { version = "53.0.0", path = "./arrow-row" }
+arrow-schema = { version = "53.0.0", path = "./arrow-schema" }
+arrow-select = { version = "53.0.0", path = "./arrow-select" }
+arrow-string = { version = "53.0.0", path = "./arrow-string" }
+parquet = { version = "53.0.0", path = "./parquet", default-features = false }
 
 chrono = { version = "0.4.34", default-features = false, features = ["clock"] }
diff --git a/arrow-array/src/cast.rs b/arrow-array/src/cast.rs
index 7b4b1d6ec..cda179b78 100644
--- a/arrow-array/src/cast.rs
+++ b/arrow-array/src/cast.rs
@@ -815,6 +815,14 @@ pub trait AsArray: private::Sealed {
         self.as_struct_opt().expect("struct array")
     }
 
+    /// Downcast this to a [`UnionArray`] returning `None` if not possible
+    fn as_union_opt(&self) -> Option<&UnionArray>;
+
+    /// Downcast this to a [`UnionArray`] panicking if not possible
+    fn as_union(&self) -> &UnionArray {
+        self.as_union_opt().expect("union array")
+    }
+
     /// Downcast this to a [`GenericListArray`] returning `None` if not possible
     fn as_list_opt<O: OffsetSizeTrait>(&self) -> Option<&GenericListArray<O>>;
 
@@ -888,6 +896,10 @@ impl AsArray for dyn Array + '_ {
         self.as_any().downcast_ref()
     }
 
+    fn as_union_opt(&self) -> Option<&UnionArray> {
+        self.as_any().downcast_ref()
+    }
+
     fn as_list_opt<O: OffsetSizeTrait>(&self) -> Option<&GenericListArray<O>> {
         self.as_any().downcast_ref()
     }
@@ -939,6 +951,10 @@ impl AsArray for ArrayRef {
         self.as_ref().as_struct_opt()
     }
 
+    fn as_union_opt(&self) -> Option<&UnionArray> {
+        self.as_any().downcast_ref()
+    }
+
     fn as_list_opt<O: OffsetSizeTrait>(&self) -> Option<&GenericListArray<O>> {
         self.as_ref().as_list_opt()
     }
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/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/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-select/src/filter.rs b/arrow-select/src/filter.rs
index d72794b3e..e07b03d1f 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -153,7 +153,10 @@ pub fn prep_null_mask_filter(filter: &BooleanArray) -> BooleanArray {
     BooleanArray::new(mask, None)
 }
 
-/// Filters an [Array], returning elements matching the filter (i.e. where the values are true).
+/// Returns a filtered `values` [Array] where the corresponding elements of
+/// `predicate` are `true`.
+///
+/// See also [`FilterBuilder`] for more control over the filtering process.
 ///
 /// # Example
 /// ```rust
@@ -166,11 +169,33 @@ pub fn prep_null_mask_filter(filter: &BooleanArray) -> BooleanArray {
 /// assert_eq!(c, &Int32Array::from(vec![5, 8]));
 /// ```
 pub fn filter(values: &dyn Array, predicate: &BooleanArray) -> Result<ArrayRef, ArrowError> {
-    let predicate = FilterBuilder::new(predicate).build();
+    let mut filter_builder = FilterBuilder::new(predicate);
+
+    if multiple_arrays(values.data_type()) {
+        // Only optimize if filtering more than one array
+        // Otherwise, the overhead of optimization can be more than the benefit
+        filter_builder = filter_builder.optimize();
+    }
+
+    let predicate = filter_builder.build();
+
     filter_array(values, &predicate)
 }
 
-/// Returns a new [RecordBatch] with arrays containing only values matching the filter.
+fn multiple_arrays(data_type: &DataType) -> bool {
+    match data_type {
+        DataType::Struct(fields) => {
+            fields.len() > 1 || fields.len() == 1 && multiple_arrays(fields[0].data_type())
+        }
+        DataType::Union(fields, UnionMode::Sparse) => !fields.is_empty(),
+        _ => false,
+    }
+}
+
+/// Returns a filtered [RecordBatch] where the corresponding elements of
+/// `predicate` are true.
+///
+/// This is the equivalent of calling [filter] on each column of the [RecordBatch].
 pub fn filter_record_batch(
     record_batch: &RecordBatch,
     predicate: &BooleanArray,
@@ -178,6 +203,7 @@ pub fn filter_record_batch(
     let mut filter_builder = FilterBuilder::new(predicate);
     if record_batch.num_columns() > 1 {
         // Only optimize if filtering more than one column
+        // Otherwise, the overhead of optimization can be more than the benefit
         filter_builder = filter_builder.optimize();
     }
     let filter = filter_builder.build();
@@ -358,6 +384,12 @@ fn filter_array(values: &dyn Array, predicate: &FilterPredicate) -> Result<Array
                 values => Ok(Arc::new(filter_dict(values, predicate))),
                 t => unimplemented!("Filter not supported for dictionary type {:?}", t)
             }
+            DataType::Struct(_) => {
+                Ok(Arc::new(filter_struct(values.as_struct(), predicate)?))
+            }
+            DataType::Union(_, UnionMode::Sparse) => {
+                Ok(Arc::new(filter_sparse_union(values.as_union(), predicate)?))
+            }
             _ => {
                 let data = values.to_data();
                 // fallback to using MutableArrayData
@@ -782,6 +814,49 @@ where
     DictionaryArray::from(unsafe { builder.build_unchecked() })
 }
 
+/// `filter` implementation for structs
+fn filter_struct(
+    array: &StructArray,
+    predicate: &FilterPredicate,
+) -> Result<StructArray, ArrowError> {
+    let columns = array
+        .columns()
+        .iter()
+        .map(|column| filter_array(column, predicate))
+        .collect::<Result<_, _>>()?;
+
+    let nulls = if let Some((null_count, nulls)) = filter_null_mask(array.nulls(), predicate) {
+        let buffer = BooleanBuffer::new(nulls, 0, predicate.count);
+
+        Some(unsafe { NullBuffer::new_unchecked(buffer, null_count) })
+    } else {
+        None
+    };
+
+    Ok(unsafe { StructArray::new_unchecked(array.fields().clone(), columns, nulls) })
+}
+
+/// `filter` implementation for sparse unions
+fn filter_sparse_union(
+    array: &UnionArray,
+    predicate: &FilterPredicate,
+) -> Result<UnionArray, ArrowError> {
+    let DataType::Union(fields, UnionMode::Sparse) = array.data_type() else {
+        unreachable!()
+    };
+
+    let type_ids = filter_primitive(&Int8Array::new(array.type_ids().clone(), None), predicate);
+
+    let children = fields
+        .iter()
+        .map(|(child_type_id, _)| filter_array(array.child(child_type_id), predicate))
+        .collect::<Result<_, _>>()?;
+
+    Ok(unsafe {
+        UnionArray::new_unchecked(fields.clone(), type_ids.into_parts().1, None, children)
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use arrow_array::builder::*;
@@ -1871,4 +1946,75 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn test_filter_struct() {
+        let predicate = BooleanArray::from(vec![true, false, true, false]);
+
+        let a = Arc::new(StringArray::from(vec!["hello", " ", "world", "!"]));
+        let a_filtered = Arc::new(StringArray::from(vec!["hello", "world"]));
+
+        let b = Arc::new(Int32Array::from(vec![5, 6, 7, 8]));
+        let b_filtered = Arc::new(Int32Array::from(vec![5, 7]));
+
+        let null_mask = NullBuffer::from(vec![true, false, false, true]);
+        let null_mask_filtered = NullBuffer::from(vec![true, false]);
+
+        let a_field = Field::new("a", DataType::Utf8, false);
+        let b_field = Field::new("b", DataType::Int32, false);
+
+        let array = StructArray::new(vec![a_field.clone()].into(), vec![a.clone()], None);
+        let expected =
+            StructArray::new(vec![a_field.clone()].into(), vec![a_filtered.clone()], None);
+
+        let result = filter(&array, &predicate).unwrap();
+
+        assert_eq!(result.to_data(), expected.to_data());
+
+        let array = StructArray::new(
+            vec![a_field.clone()].into(),
+            vec![a.clone()],
+            Some(null_mask.clone()),
+        );
+        let expected = StructArray::new(
+            vec![a_field.clone()].into(),
+            vec![a_filtered.clone()],
+            Some(null_mask_filtered.clone()),
+        );
+
+        let result = filter(&array, &predicate).unwrap();
+
+        assert_eq!(result.to_data(), expected.to_data());
+
+        let array = StructArray::new(
+            vec![a_field.clone(), b_field.clone()].into(),
+            vec![a.clone(), b.clone()],
+            None,
+        );
+        let expected = StructArray::new(
+            vec![a_field.clone(), b_field.clone()].into(),
+            vec![a_filtered.clone(), b_filtered.clone()],
+            None,
+        );
+
+        let result = filter(&array, &predicate).unwrap();
+
+        assert_eq!(result.to_data(), expected.to_data());
+
+        let array = StructArray::new(
+            vec![a_field.clone(), b_field.clone()].into(),
+            vec![a.clone(), b.clone()],
+            Some(null_mask.clone()),
+        );
+
+        let expected = StructArray::new(
+            vec![a_field.clone(), b_field.clone()].into(),
+            vec![a_filtered.clone(), b_filtered.clone()],
+            Some(null_mask_filtered.clone()),
+        );
+
+        let result = filter(&array, &predicate).unwrap();
+
+        assert_eq!(result.to_data(), expected.to_data());
+    }
 }
diff --git a/arrow/src/pyarrow.rs b/arrow/src/pyarrow.rs
index 43cdb4fe0..336398cbf 100644
--- a/arrow/src/pyarrow.rs
+++ b/arrow/src/pyarrow.rs
@@ -59,7 +59,7 @@ use std::convert::{From, TryFrom};
 use std::ptr::{addr_of, addr_of_mut};
 use std::sync::Arc;
 
-use arrow_array::{RecordBatchIterator, RecordBatchReader, StructArray};
+use arrow_array::{RecordBatchIterator, RecordBatchOptions, RecordBatchReader, StructArray};
 use pyo3::exceptions::{PyTypeError, PyValueError};
 use pyo3::ffi::Py_uintptr_t;
 use pyo3::import_exception;
@@ -361,6 +361,7 @@ impl FromPyArrow for RecordBatch {
                     "Expected Struct type from __arrow_c_array.",
                 ));
             }
+            let options = RecordBatchOptions::default().with_row_count(Some(array_data.len()));
             let array = StructArray::from(array_data);
             // StructArray does not embed metadata from schema. We need to override
             // the output schema with the schema from the capsule.
@@ -371,7 +372,7 @@ impl FromPyArrow for RecordBatch {
                 0,
                 "Cannot convert nullable StructArray to RecordBatch, see StructArray documentation"
             );
-            return RecordBatch::try_new(schema, columns).map_err(to_py_err);
+            return RecordBatch::try_new_with_options(schema, columns, &options).map_err(to_py_err);
         }
 
         validate_class("RecordBatch", value)?;
@@ -386,7 +387,14 @@ impl FromPyArrow for RecordBatch {
             .map(|a| Ok(make_array(ArrayData::from_pyarrow_bound(&a)?)))
             .collect::<PyResult<_>>()?;
 
-        let batch = RecordBatch::try_new(schema, arrays).map_err(to_py_err)?;
+        let row_count = value
+            .getattr("num_rows")
+            .ok()
+            .and_then(|x| x.extract().ok());
+        let options = RecordBatchOptions::default().with_row_count(row_count);
+
+        let batch =
+            RecordBatch::try_new_with_options(schema, arrays, &options).map_err(to_py_err)?;
         Ok(batch)
     }
 }
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/dev/release/update_change_log.sh b/dev/release/update_change_log.sh
index 20d7e60aa..5c4fcb53a 100755
--- a/dev/release/update_change_log.sh
+++ b/dev/release/update_change_log.sh
@@ -29,8 +29,8 @@
 
 set -e
 
-SINCE_TAG="52.1.0"
-FUTURE_RELEASE="52.2.0"
+SINCE_TAG="52.2.0"
+FUTURE_RELEASE="53.0.0"
 
 SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 SOURCE_TOP_DIR="$(cd "${SOURCE_DIR}/../../" && pwd)"
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/azure/client.rs b/object_store/src/azure/client.rs
index b5e82c2a8..049905155 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -226,11 +226,16 @@ impl<'a> PutRequest<'a> {
 
     async fn send(self) -> Result<Response> {
         let credential = self.config.get_credential().await?;
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         let response = self
             .builder
             .header(CONTENT_LENGTH, self.payload.content_length())
             .with_azure_authorization(&credential, &self.config.account)
             .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
             .idempotent(self.idempotent)
             .payload(Some(self.payload))
             .send()
@@ -356,12 +361,18 @@ impl AzureClient {
         let credential = self.get_credential().await?;
         let url = self.config.path_url(path);
 
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         self.client
             .request(Method::DELETE, url)
             .query(query)
             .header(&DELETE_SNAPSHOTS, "include")
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
+            .send()
             .await
             .context(DeleteRequestSnafu {
                 path: path.as_ref(),
@@ -392,9 +403,14 @@ impl AzureClient {
             builder = builder.header(IF_NONE_MATCH, "*");
         }
 
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         builder
             .with_azure_authorization(&credential, &self.config.account)
             .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
             .idempotent(overwrite)
             .send()
             .await
@@ -423,6 +439,10 @@ impl AzureClient {
         ));
         body.push_str("</KeyInfo>");
 
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         let response = self
             .client
             .request(Method::POST, url)
@@ -430,6 +450,7 @@ impl AzureClient {
             .query(&[("restype", "service"), ("comp", "userdelegationkey")])
             .with_azure_authorization(&credential, &self.config.account)
             .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
             .idempotent(true)
             .send()
             .await
@@ -482,12 +503,18 @@ impl AzureClient {
     pub async fn get_blob_tagging(&self, path: &Path) -> Result<Response> {
         let credential = self.get_credential().await?;
         let url = self.config.path_url(path);
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         let response = self
             .client
             .request(Method::GET, url)
             .query(&[("comp", "tags")])
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
+            .send()
             .await
             .context(GetRequestSnafu {
                 path: path.as_ref(),
@@ -536,10 +563,16 @@ impl GetClient for AzureClient {
             builder = builder.query(&[("versionid", v)])
         }
 
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         let response = builder
             .with_get_options(options)
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
+            .send()
             .await
             .context(GetRequestSnafu {
                 path: path.as_ref(),
@@ -590,12 +623,18 @@ impl ListClient for AzureClient {
             query.push(("marker", token))
         }
 
+        let sensitive = credential
+            .as_deref()
+            .map(|c| c.sensitive_request())
+            .unwrap_or_default();
         let response = self
             .client
             .request(Method::GET, url)
             .query(&query)
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .retryable(&self.config.retry_config)
+            .sensitive(sensitive)
+            .send()
             .await
             .context(ListRequestSnafu)?
             .bytes()
diff --git a/object_store/src/azure/credential.rs b/object_store/src/azure/credential.rs
index c8212a929..7808c7c4a 100644
--- a/object_store/src/azure/credential.rs
+++ b/object_store/src/azure/credential.rs
@@ -130,6 +130,18 @@ pub enum AzureCredential {
     BearerToken(String),
 }
 
+impl AzureCredential {
+    /// Determines if the credential requires the request be treated as sensitive
+    pub fn sensitive_request(&self) -> bool {
+        match self {
+            Self::AccessKey(_) => false,
+            Self::BearerToken(_) => false,
+            // SAS tokens are sent as query parameters in the url
+            Self::SASToken(_) => true,
+        }
+    }
+}
+
 /// A list of known Azure authority hosts
 pub mod authority_hosts {
     /// China-based Azure Authority Host
diff --git a/object_store/src/client/backoff.rs b/object_store/src/client/backoff.rs
index e01589102..a1fa26cde 100644
--- a/object_store/src/client/backoff.rs
+++ b/object_store/src/client/backoff.rs
@@ -18,7 +18,12 @@
 use rand::prelude::*;
 use std::time::Duration;
 
-/// Exponential backoff with jitter
+/// Exponential backoff with decorrelated jitter algorithm
+///
+/// The first backoff will always be `init_backoff`.
+///
+/// Subsequent backoffs will pick a random value between `init_backoff` and
+/// `base * previous` where `previous` is the duration of the previous backoff
 ///
 /// See <https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/>
 #[allow(missing_copy_implementations)]
@@ -28,7 +33,7 @@ pub struct BackoffConfig {
     pub init_backoff: Duration,
     /// The maximum backoff duration
     pub max_backoff: Duration,
-    /// The base of the exponential to use
+    /// The multiplier to use for the next backoff duration
     pub base: f64,
 }
 
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/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/mod.rs b/parquet/src/arrow/mod.rs
index 5c3689143..afe7ed1eb 100644
--- a/parquet/src/arrow/mod.rs
+++ b/parquet/src/arrow/mod.rs
@@ -156,7 +156,7 @@ pub const PARQUET_FIELD_ID_META_KEY: &str = "PARQUET:field_id";
 /// For non-nested schemas, i.e. those containing only primitive columns, the root
 /// and leaves are the same
 ///
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ProjectionMask {
     /// If present a leaf column should be included if the value at
     /// the corresponding index is true
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 606ad462d..e0d3abed8 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -756,8 +756,8 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
         if null_page && self.column_index_builder.valid() {
             self.column_index_builder.append(
                 null_page,
-                vec![0; 1],
-                vec![0; 1],
+                vec![],
+                vec![],
                 self.page_metrics.num_page_nulls as i64,
             );
         } else if self.column_index_builder.valid() {
@@ -2668,6 +2668,32 @@ mod tests {
         ),);
     }
 
+    #[test]
+    fn test_column_index_with_null_pages() {
+        // write a single page of all nulls
+        let page_writer = get_test_page_writer();
+        let props = Default::default();
+        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 1, 0, props);
+        writer.write_batch(&[], Some(&[0, 0, 0, 0]), None).unwrap();
+
+        let r = writer.close().unwrap();
+        assert!(r.column_index.is_some());
+        let col_idx = r.column_index.unwrap();
+        // null_pages should be true for page 0
+        assert!(col_idx.null_pages[0]);
+        // min and max should be empty byte arrays
+        assert_eq!(col_idx.min_values[0].len(), 0);
+        assert_eq!(col_idx.max_values[0].len(), 0);
+        // null_counts should be defined and be 4 for page 0
+        assert!(col_idx.null_counts.is_some());
+        assert_eq!(col_idx.null_counts.as_ref().unwrap()[0], 4);
+        // there is no repetition so rep histogram should be absent
+        assert!(col_idx.repetition_level_histograms.is_none());
+        // definition_level_histogram should be present and should be 0:4, 1:0
+        assert!(col_idx.definition_level_histograms.is_some());
+        assert_eq!(col_idx.definition_level_histograms.unwrap(), &[4, 0]);
+    }
+
     #[test]
     fn test_column_offset_index_metadata() {
         // write data
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/file/metadata/writer.rs b/parquet/src/file/metadata/writer.rs
index 3573e3020..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);
                 }
             }
         }
@@ -450,19 +442,57 @@ mod tests {
             true,
         )]));
 
-        let a: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None, Some(2)]));
+        // build row groups / pages that exercise different combinations of nulls and values
+        // note that below we set the row group and page sizes to 4 and 2 respectively
+        // so that these "groupings" make sense
+        let a: ArrayRef = Arc::new(Int32Array::from(vec![
+            // a row group that has all values
+            Some(i32::MIN),
+            Some(-1),
+            Some(1),
+            Some(i32::MAX),
+            // a row group with a page of all nulls and a page of all values
+            None,
+            None,
+            Some(2),
+            Some(3),
+            // a row group that has all null pages
+            None,
+            None,
+            None,
+            None,
+            // a row group having 1 page with all values and 1 page with some nulls
+            Some(4),
+            Some(5),
+            None,
+            Some(6),
+            // a row group having 1 page with all nulls and 1 page with some nulls
+            None,
+            None,
+            Some(7),
+            None,
+            // a row group having all pages with some nulls
+            None,
+            Some(8),
+            Some(9),
+            None,
+        ]));
 
         let batch = RecordBatch::try_from_iter(vec![("a", a)]).unwrap();
 
-        let writer_props = match write_page_index {
-            true => WriterProperties::builder()
-                .set_statistics_enabled(EnabledStatistics::Page)
-                .build(),
-            false => WriterProperties::builder()
-                .set_statistics_enabled(EnabledStatistics::Chunk)
-                .build(),
+        let writer_props_builder = match write_page_index {
+            true => WriterProperties::builder().set_statistics_enabled(EnabledStatistics::Page),
+            false => WriterProperties::builder().set_statistics_enabled(EnabledStatistics::Chunk),
         };
 
+        // tune the size or pages to the data above
+        // to make sure we exercise code paths where all items in a page are null, etc.
+        let writer_props = writer_props_builder
+            .set_max_row_group_size(4)
+            .set_data_page_row_count_limit(2)
+            .set_write_batch_size(2)
+            .build();
+
         let mut writer = ArrowWriter::try_new(&mut buf, schema, Some(writer_props)).unwrap();
         writer.write(&batch).unwrap();
         writer.close().unwrap();
@@ -486,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;
@@ -610,6 +639,29 @@ mod tests {
             decoded_metadata.num_row_groups()
         );
 
+        // check that the mins and maxes are what we expect for each page
+        // also indirectly checking that the pages were written out as we expected them to be laid out
+        // (if they're not, or something gets refactored in the future that breaks that assumption,
+        // this test may have to drop down to a lower level and create metadata directly instead of relying on
+        // writing an entire file)
+        let column_indexes = metadata.metadata.column_index().unwrap();
+        assert_eq!(column_indexes.len(), 6);
+        // make sure each row group has 2 pages by checking the first column
+        // page counts for each column for each row group, should all be the same and there should be
+        // 12 pages in total across 6 row groups / 1 column
+        let mut page_counts = vec![];
+        for row_group in column_indexes {
+            for column in row_group {
+                match column {
+                    Index::INT32(column_index) => {
+                        page_counts.push(column_index.indexes.len());
+                    }
+                    _ => panic!("unexpected column index type"),
+                }
+            }
+        }
+        assert_eq!(page_counts, vec![2; 6]);
+
         metadata
             .metadata
             .row_groups()
diff --git a/parquet/src/file/page_index/index.rs b/parquet/src/file/page_index/index.rs
index 662ba4562..2f30abead 100644
--- a/parquet/src/file/page_index/index.rs
+++ b/parquet/src/file/page_index/index.rs
@@ -230,16 +230,14 @@ impl<T: ParquetValueType> NativeIndex<T> {
         let min_values = self
             .indexes
             .iter()
-            .map(|x| x.min_bytes().map(|x| x.to_vec()))
-            .collect::<Option<Vec<_>>>()
-            .unwrap_or_else(|| vec![vec![]; self.indexes.len()]);
+            .map(|x| x.min_bytes().unwrap_or(&[]).to_vec())
+            .collect::<Vec<_>>();
 
         let max_values = self
             .indexes
             .iter()
-            .map(|x| x.max_bytes().map(|x| x.to_vec()))
-            .collect::<Option<Vec<_>>>()
-            .unwrap_or_else(|| vec![vec![]; self.indexes.len()]);
+            .map(|x| x.max_bytes().unwrap_or(&[]).to_vec())
+            .collect::<Vec<_>>();
 
         let null_counts = self
             .indexes
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_derive/src/lib.rs b/parquet_derive/src/lib.rs
index 16b6a6699..9c93e2cca 100644
--- a/parquet_derive/src/lib.rs
+++ b/parquet_derive/src/lib.rs
@@ -146,10 +146,10 @@ pub fn parquet_record_writer(input: proc_macro::TokenStream) -> proc_macro::Toke
 /// Derive flat, simple RecordReader implementations. Works by parsing
 /// a struct tagged with `#[derive(ParquetRecordReader)]` and emitting
 /// the correct writing code for each field of the struct. Column readers
-/// are generated in the order they are defined.
+/// are generated by matching names in the schema to the names in the struct.
 ///
-/// It is up to the programmer to keep the order of the struct
-/// fields lined up with the schema.
+/// It is up to the programmer to ensure the names in the struct
+/// fields line up with the schema.
 ///
 /// Example:
 ///
@@ -189,7 +189,6 @@ pub fn parquet_record_reader(input: proc_macro::TokenStream) -> proc_macro::Toke
     let field_names: Vec<_> = fields.iter().map(|f| f.ident.clone()).collect();
     let reader_snippets: Vec<proc_macro2::TokenStream> =
         field_infos.iter().map(|x| x.reader_snippet()).collect();
-    let i: Vec<_> = (0..reader_snippets.len()).collect();
 
     let derived_for = input.ident;
     let generics = input.generics;
@@ -206,6 +205,12 @@ pub fn parquet_record_reader(input: proc_macro::TokenStream) -> proc_macro::Toke
 
         let mut row_group_reader = row_group_reader;
 
+        // key: parquet file column name, value: column index
+        let mut name_to_index = std::collections::HashMap::new();
+        for (idx, col) in row_group_reader.metadata().schema_descr().columns().iter().enumerate() {
+            name_to_index.insert(col.name().to_string(), idx);
+        }
+
         for _ in 0..num_records {
           self.push(#derived_for {
             #(
@@ -218,7 +223,14 @@ pub fn parquet_record_reader(input: proc_macro::TokenStream) -> proc_macro::Toke
 
         #(
           {
-              if let Ok(mut column_reader) = row_group_reader.get_column_reader(#i) {
+              let idx: usize = match name_to_index.get(stringify!(#field_names)) {
+                Some(&col_idx) => col_idx,
+                None => {
+                  let error_msg = format!("column name '{}' is not found in parquet file!", stringify!(#field_names));
+                  return Err(::parquet::errors::ParquetError::General(error_msg));
+                }
+              };
+              if let Ok(mut column_reader) = row_group_reader.get_column_reader(idx) {
                   #reader_snippets
               } else {
                   return Err(::parquet::errors::ParquetError::General("Failed to get next column".into()))


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/arrow-pyarrow-integration-testing/tests/test_sql.py b/arrow-pyarrow-integration-testing/tests/test_sql.py
index 5320d0a53..3b46d5729 100644
--- a/arrow-pyarrow-integration-testing/tests/test_sql.py
+++ b/arrow-pyarrow-integration-testing/tests/test_sql.py
@@ -476,6 +476,29 @@ def test_tensor_array():
 
     del b
 
+
+def test_empty_recordbatch_with_row_count():
+    """
+    A pyarrow.RecordBatch with no columns but with `num_rows` set.
+
+    `datafusion-python` gets this as the result of a `count(*)` query.  
+    """
+
+    # Create an empty schema with no fields
+    batch = pa.RecordBatch.from_pydict({"a": [1, 2, 3, 4]}).select([])
+    num_rows = 4
+    assert batch.num_rows == num_rows
+    assert batch.num_columns == 0
+
+    b = rust.round_trip_record_batch(batch)
+    assert b == batch
+    assert b.schema == batch.schema
+    assert b.schema.metadata == batch.schema.metadata
+
+    assert b.num_rows == batch.num_rows
+    
+    del b
+
 def test_record_batch_reader():
     """
     Python -> Rust -> Python
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-testing b/parquet-testing
index cb7a96741..50af3d8ce 160000
--- a/parquet-testing
+++ b/parquet-testing
@@ -1,1 +1,1 @@
-Subproject commit cb7a9674142c137367bf75a01b79c6e214a73199
+Subproject commit 50af3d8ce206990d81014b1862e5ce7380dc3e08
diff --git a/parquet/tests/arrow_reader/bad_data.rs b/parquet/tests/arrow_reader/bad_data.rs
index cbd5d4d3b..a73864070 100644
--- a/parquet/tests/arrow_reader/bad_data.rs
+++ b/parquet/tests/arrow_reader/bad_data.rs
@@ -27,6 +27,7 @@ static KNOWN_FILES: &[&str] = &[
     "PARQUET-1481.parquet",
     "ARROW-GH-41317.parquet",
     "ARROW-GH-41321.parquet",
+    "ARROW-GH-43605.parquet",
     "ARROW-RS-GH-6229-DICTHEADER.parquet",
     "ARROW-RS-GH-6229-LEVELS.parquet",
     "README.md",
diff --git a/parquet_derive_test/src/lib.rs b/parquet_derive_test/src/lib.rs
index e7c7896cb..2cd69d03d 100644
--- a/parquet_derive_test/src/lib.rs
+++ b/parquet_derive_test/src/lib.rs
@@ -73,9 +73,9 @@ struct APartiallyCompleteRecord {
 struct APartiallyOptionalRecord {
     pub bool: bool,
     pub string: String,
-    pub maybe_i16: Option<i16>,
-    pub maybe_i32: Option<i32>,
-    pub maybe_u64: Option<u64>,
+    pub i16: Option<i16>,
+    pub i32: Option<i32>,
+    pub u64: Option<u64>,
     pub isize: isize,
     pub float: f32,
     pub double: f64,
@@ -85,6 +85,22 @@ struct APartiallyOptionalRecord {
     pub byte_vec: Vec<u8>,
 }
 
+// This struct removes several fields from the "APartiallyCompleteRecord",
+// and it shuffles the fields.
+// we should still be able to load it from APartiallyCompleteRecord parquet file
+#[derive(PartialEq, ParquetRecordReader, Debug)]
+struct APrunedRecord {
+    pub bool: bool,
+    pub string: String,
+    pub byte_vec: Vec<u8>,
+    pub float: f32,
+    pub double: f64,
+    pub i16: i16,
+    pub i32: i32,
+    pub u64: u64,
+    pub isize: isize,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -240,12 +256,12 @@ mod tests {
     #[test]
     fn test_parquet_derive_read_optional_but_valid_column() {
         let file = get_temp_file("test_parquet_derive_read_optional", &[]);
-        let drs: Vec<APartiallyOptionalRecord> = vec![APartiallyOptionalRecord {
+        let drs = vec![APartiallyOptionalRecord {
             bool: true,
             string: "a string".into(),
-            maybe_i16: Some(-45),
-            maybe_i32: Some(456),
-            maybe_u64: Some(4563424),
+            i16: Some(-45),
+            i32: Some(456),
+            u64: Some(4563424),
             isize: -365,
             float: 3.5,
             double: f64::NAN,
@@ -273,9 +289,57 @@ mod tests {
         let mut row_group = reader.get_row_group(0).unwrap();
         out.read_from_row_group(&mut *row_group, 1).unwrap();
 
-        assert_eq!(drs[0].maybe_i16.unwrap(), out[0].i16);
-        assert_eq!(drs[0].maybe_i32.unwrap(), out[0].i32);
-        assert_eq!(drs[0].maybe_u64.unwrap(), out[0].u64);
+        assert_eq!(drs[0].i16.unwrap(), out[0].i16);
+        assert_eq!(drs[0].i32.unwrap(), out[0].i32);
+        assert_eq!(drs[0].u64.unwrap(), out[0].u64);
+    }
+
+    #[test]
+    fn test_parquet_derive_read_pruned_and_shuffled_columns() {
+        let file = get_temp_file("test_parquet_derive_read_pruned", &[]);
+        let drs = vec![APartiallyCompleteRecord {
+            bool: true,
+            string: "a string".into(),
+            i16: -45,
+            i32: 456,
+            u64: 4563424,
+            isize: -365,
+            float: 3.5,
+            double: f64::NAN,
+            now: chrono::Utc::now().naive_local(),
+            date: chrono::naive::NaiveDate::from_ymd_opt(2015, 3, 14).unwrap(),
+            uuid: uuid::Uuid::new_v4(),
+            byte_vec: vec![0x65, 0x66, 0x67],
+        }];
+
+        let generated_schema = drs.as_slice().schema().unwrap();
+
+        let props = Default::default();
+        let mut writer =
+            SerializedFileWriter::new(file.try_clone().unwrap(), generated_schema, props).unwrap();
+
+        let mut row_group = writer.next_row_group().unwrap();
+        drs.as_slice().write_to_row_group(&mut row_group).unwrap();
+        row_group.close().unwrap();
+        writer.close().unwrap();
+
+        use parquet::file::{reader::FileReader, serialized_reader::SerializedFileReader};
+        let reader = SerializedFileReader::new(file).unwrap();
+        let mut out: Vec<APrunedRecord> = Vec::new();
+
+        let mut row_group = reader.get_row_group(0).unwrap();
+        out.read_from_row_group(&mut *row_group, 1).unwrap();
+
+        assert_eq!(drs[0].bool, out[0].bool);
+        assert_eq!(drs[0].string, out[0].string);
+        assert_eq!(drs[0].byte_vec, out[0].byte_vec);
+        assert_eq!(drs[0].float, out[0].float);
+        assert!(drs[0].double.is_nan());
+        assert!(out[0].double.is_nan());
+        assert_eq!(drs[0].i16, out[0].i16);
+        assert_eq!(drs[0].i32, out[0].i32);
+        assert_eq!(drs[0].u64, out[0].u64);
+        assert_eq!(drs[0].isize, out[0].isize);
     }
 
     /// Returns file handle for a temp file in 'target' directory with a provided content