Repository: apache/arrow-rs
Pull request #6236 (2024-08-13 10:07:06)

docs: Add parquet_opendal in related projects
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes https://github.com/apache/arrow-rs/issues/6235

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

Covered in https://github.com/apache/arrow-rs/issues/6235

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

Only docs change.

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->

No.


Patch:
diff --git a/.github/workflows/take.yml b/.github/workflows/take.yml
new file mode 100644
index 000000000..86dc190ad
--- /dev/null
+++ b/.github/workflows/take.yml
@@ -0,0 +1,41 @@
+# 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.
+
+name: Assign the issue via a `take` comment
+on:
+  issue_comment:
+    types: created
+
+permissions:
+  issues: write
+
+jobs:
+  issue_assign:
+    runs-on: ubuntu-latest
+    if: (!github.event.issue.pull_request) && github.event.comment.body == 'take'
+    concurrency:
+      group: ${{ github.actor }}-issue-assign
+    steps:
+      - run: |
+          CODE=$(curl -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" -LI https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.issue.number }}/assignees/${{ github.event.comment.user.login }} -o /dev/null -w '%{http_code}\n' -s)
+          if [ "$CODE" -eq "204" ]
+          then
+            echo "Assigning issue ${{ github.event.issue.number }} to ${{ github.event.comment.user.login }}"
+            curl -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" -d '{"assignees": ["${{ github.event.comment.user.login }}"]}' https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.issue.number }}/assignees
+          else
+            echo "Cannot assign issue ${{ github.event.issue.number }} to ${{ github.event.comment.user.login }}"
+          fi
\ No newline at end of file
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index e0d792f5e..a53604d0f 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -25,7 +25,41 @@ We welcome and encourage contributions of all kinds, such as:
 2. Documentation improvements
 3. Code (PR or PR Review)
 
-In addition to submitting new PRs, we have a healthy tradition of community members helping review each other's PRs. Doing so is a great way to help the community as well as get more familiar with Rust and the relevant codebases.
+In addition to submitting new PRs, we have a healthy tradition of community
+members helping review each other's PRs. Doing so is a great way to help the
+community as well as get more familiar with Rust and the relevant codebases.
+
+## Finding and Creating Issues to Work On
+
+You can find a curated [good-first-issue] list to help you get started.
+
+Arrow-rs is an open contribution project, and thus there is no particular
+project imposed deadline for completing any issue or any restriction on who can
+work on an issue, nor how many people can work on an issue at the same time.
+
+Contributors drive the project forward based on their own priorities and
+interests and thus you are free to work on any issue that interests you.
+
+If someone is already working on an issue that you want or need but hasn't
+been able to finish it yet, you should feel free to work on it as well. In
+general it is both polite and will help avoid unnecessary duplication of work if
+you leave a note on an issue when you start working on it.
+
+If you want to work on an issue which is not already assigned to someone else
+and there are no comment indicating that someone is already working on that
+issue then you can assign the issue to yourself by submitting a single word
+comment `take`. This will assign the issue to yourself. However, if you are
+unable to make progress you should unassign the issue by using the `unassign me`
+link at the top of the issue page (and ask for help if are stuck) so that
+someone else can get involved in the work.
+
+If you plan to work on a new feature that doesn't have an existing ticket, it is
+a good idea to open a ticket to discuss the feature. Advanced discussion often
+helps avoid wasted effort by determining early if the feature is a good fit for
+Arrow-rs before too much time is invested. It also often helps to discuss your
+ideas with the community to get feedback on implementation.
+
+[good-first-issue]: https://github.com/apache/arrow-rs/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22
 
 ## Developer's guide to Arrow Rust
 
diff --git a/LICENSE.txt b/LICENSE.txt
index d74c6b599..d64569567 100644
--- a/LICENSE.txt
+++ b/LICENSE.txt
@@ -200,13 +200,3 @@
    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.
-
-
-This project includes code from Apache Aurora.
-
-* dev/release/{release,changelog,release-candidate} are based on the scripts from
-  Apache Aurora
-
-Copyright: 2016 The Apache Software Foundation.
-Home page: https://aurora.apache.org/
-License: http://www.apache.org/licenses/LICENSE-2.0
diff --git a/README.md b/README.md
index 944e8562b..a52549084 100644
--- a/README.md
+++ b/README.md
@@ -92,12 +92,15 @@ There are several related crates in different repositories
 | [`datafusion`]           | In-memory query engine with SQL support     | [(README)][datafusion-readme]           |
 | [`ballista`]             | Distributed query execution                 | [(README)][ballista-readme]             |
 | [`object_store_opendal`] | Use [`opendal`] as [`object_store`] backend | [(README)][object_store_opendal-readme] |
+| [`parquet_opendal`]      | Use [`opendal`] for [`parquet`] Arrow IO    | [(README)][parquet_opendal-readme]      |
 
 [`datafusion`]: https://crates.io/crates/datafusion
 [`ballista`]: https://crates.io/crates/ballista
 [`object_store_opendal`]: https://crates.io/crates/object_store_opendal
 [`opendal`]: https://crates.io/crates/opendal
 [object_store_opendal-readme]: https://github.com/apache/opendal/blob/main/integrations/object_store/README.md
+[`parquet_opendal`]: https://crates.io/crates/parquet_opendal
+[parquet_opendal-readme]: https://github.com/apache/opendal/blob/main/integrations/parquet/README.md
 
 Collectively, these crates support a wider array of functionality for analytic computations in Rust.
 
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index a9aed9531..42f945838 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -34,13 +34,25 @@ use super::ByteArrayType;
 
 /// [Variable-size Binary View Layout]: An array of variable length bytes view arrays.
 ///
-/// Different than [`crate::GenericByteArray`] as it stores both an offset and length
-/// meaning that take / filter operations can be implemented without copying the underlying data.
+/// [Variable-size Binary View Layout]: https://arrow.apache.org/docs/format/Columnar.html#variable-size-binary-view-layout
+///
+/// This is different from [`GenericByteArray`] as it stores both an offset and
+/// length meaning that take / filter operations can be implemented without
+/// copying the underlying data. In addition, it stores an inlined prefix which
+/// can be used to speed up comparisons.
+///
+/// # See Also
 ///
 /// See [`StringViewArray`] for storing utf8 encoded string data and
 /// [`BinaryViewArray`] for storing bytes.
 ///
-/// [Variable-size Binary View Layout]: https://arrow.apache.org/docs/format/Columnar.html#variable-size-binary-view-layout
+/// # Notes
+///
+/// Comparing two `GenericByteViewArray` using PartialEq compares by structure,
+/// not by value. as there are many different buffer layouts to represent the
+/// same data (e.g. different offsets, different buffer sizes, etc).
+///
+/// # Layout
 ///
 /// A `GenericByteViewArray` stores variable length byte strings. An array of
 /// `N` elements is stored as `N` fixed length "views" and a variable number
@@ -95,7 +107,6 @@ use super::ByteArrayType;
 ///                                                                   buffer 0 │...│
 ///                                                                            └───┘
 /// ```
-/// [`GenericByteArray`]: crate::array::GenericByteArray
 pub struct GenericByteViewArray<T: ByteViewType + ?Sized> {
     data_type: DataType,
     views: ScalarBuffer<u128>,
@@ -116,6 +127,16 @@ impl<T: ByteViewType + ?Sized> Clone for GenericByteViewArray<T> {
     }
 }
 
+// PartialEq
+impl<T: ByteViewType + ?Sized> PartialEq for GenericByteViewArray<T> {
+    fn eq(&self, other: &Self) -> bool {
+        other.data_type.eq(&self.data_type)
+            && other.views.eq(&self.views)
+            && other.buffers.eq(&self.buffers)
+            && other.nulls.eq(&self.nulls)
+    }
+}
+
 impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
     /// Create a new [`GenericByteViewArray`] from the provided parts, panicking on failure
     ///
@@ -870,4 +891,29 @@ mod tests {
         check_gc(&array.slice(2, 2));
         check_gc(&array.slice(3, 1));
     }
+
+    #[test]
+    fn test_eq() {
+        let test_data = [
+            Some("longer than 12 bytes"),
+            None,
+            Some("short"),
+            Some("again, this is longer than 12 bytes"),
+        ];
+
+        let array1 = {
+            let mut builder = StringViewBuilder::new().with_fixed_block_size(8);
+            test_data.into_iter().for_each(|v| builder.append_option(v));
+            builder.finish()
+        };
+        let array2 = {
+            // create a new array with the same data but different layout
+            let mut builder = StringViewBuilder::new().with_fixed_block_size(100);
+            test_data.into_iter().for_each(|v| builder.append_option(v));
+            builder.finish()
+        };
+        assert_eq!(array1, array1.clone());
+        assert_eq!(array2, array2.clone());
+        assert_ne!(array1, array2);
+    }
 }
diff --git a/arrow-array/src/array/primitive_array.rs b/arrow-array/src/array/primitive_array.rs
index 70a8ceaef..db14845b0 100644
--- a/arrow-array/src/array/primitive_array.rs
+++ b/arrow-array/src/array/primitive_array.rs
@@ -713,6 +713,20 @@ impl<T: ArrowPrimitiveType> PrimitiveArray<T> {
         }
     }
 
+    /// Creates a PrimitiveArray based on an iterator of values with provided nulls
+    pub fn from_iter_values_with_nulls<I: IntoIterator<Item = T::Native>>(
+        iter: I,
+        nulls: Option<NullBuffer>,
+    ) -> Self {
+        let val_buf: Buffer = iter.into_iter().collect();
+        let len = val_buf.len() / std::mem::size_of::<T::Native>();
+        Self {
+            data_type: T::DATA_TYPE,
+            values: ScalarBuffer::new(val_buf, 0, len),
+            nulls,
+        }
+    }
+
     /// Creates a PrimitiveArray based on a constant value with `count` elements
     pub fn from_value(value: T::Native, count: usize) -> Self {
         unsafe {
diff --git a/arrow-array/src/builder/generic_bytes_view_builder.rs b/arrow-array/src/builder/generic_bytes_view_builder.rs
index 4f19204b8..deaf447d0 100644
--- a/arrow-array/src/builder/generic_bytes_view_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_view_builder.rs
@@ -60,12 +60,14 @@ impl BlockSizeGrowthStrategy {
 /// A [`GenericByteViewArray`] consists of a list of data blocks containing string data,
 /// and a list of views into those buffers.
 ///
+/// See examples on [`StringViewBuilder`] and [`BinaryViewBuilder`]
+///
 /// This builder can be used in two ways
 ///
 /// # Append Values
 ///
 /// To avoid bump allocating, this builder allocates data in fixed size blocks, configurable
-/// using [`GenericByteViewBuilder::with_block_size`]. [`GenericByteViewBuilder::append_value`]
+/// using [`GenericByteViewBuilder::with_fixed_block_size`]. [`GenericByteViewBuilder::append_value`]
 /// writes values larger than 12 bytes to the current in-progress block, with values smaller
 /// than 12 bytes inlined into the views. If a value is appended that will not fit in the
 /// in-progress block, it will be closed, and a new block of sufficient size allocated
@@ -462,12 +464,43 @@ impl<T: ByteViewType + ?Sized, V: AsRef<T::Native>> Extend<Option<V>>
 ///
 /// Values can be appended using [`GenericByteViewBuilder::append_value`], and nulls with
 /// [`GenericByteViewBuilder::append_null`] as normal.
+///
+/// # Example
+/// ```
+/// # use arrow_array::builder::StringViewBuilder;
+/// # use arrow_array::StringViewArray;
+/// let mut builder = StringViewBuilder::new();
+/// builder.append_value("hello");
+/// builder.append_null();
+/// builder.append_value("world");
+/// let array = builder.finish();
+///
+/// let expected = vec![Some("hello"), None, Some("world")];
+/// let actual: Vec<_> = array.iter().collect();
+/// assert_eq!(expected, actual);
+/// ```
 pub type StringViewBuilder = GenericByteViewBuilder<StringViewType>;
 
 ///  Array builder for [`BinaryViewArray`][crate::BinaryViewArray]
 ///
 /// Values can be appended using [`GenericByteViewBuilder::append_value`], and nulls with
 /// [`GenericByteViewBuilder::append_null`] as normal.
+///
+/// # Example
+/// ```
+/// # use arrow_array::builder::BinaryViewBuilder;
+/// use arrow_array::BinaryViewArray;
+/// let mut builder = BinaryViewBuilder::new();
+/// builder.append_value("hello");
+/// builder.append_null();
+/// builder.append_value("world");
+/// let array = builder.finish();
+///
+/// let expected: Vec<Option<&[u8]>> = vec![Some(b"hello"), None, Some(b"world")];
+/// let actual: Vec<_> = array.iter().collect();
+/// assert_eq!(expected, actual);
+/// ```
+///
 pub type BinaryViewBuilder = GenericByteViewBuilder<BinaryViewType>;
 
 /// Creates a view from a fixed length input (the compiler can generate
diff --git a/arrow-array/src/types.rs b/arrow-array/src/types.rs
index 86298cf3e..550d1aadf 100644
--- a/arrow-array/src/types.rs
+++ b/arrow-array/src/types.rs
@@ -101,163 +101,165 @@ make_type!(
     Int16Type,
     i16,
     DataType::Int16,
-    "A signed 16-bit integer type."
+    "Signed 16-bit integer type."
 );
 make_type!(
     Int32Type,
     i32,
     DataType::Int32,
-    "A signed 32-bit integer type."
+    "Signed 32-bit integer type."
 );
 make_type!(
     Int64Type,
     i64,
     DataType::Int64,
-    "A signed 64-bit integer type."
+    "Signed 64-bit integer type."
 );
 make_type!(
     UInt8Type,
     u8,
     DataType::UInt8,
-    "An unsigned 8-bit integer type."
+    "Unsigned 8-bit integer type."
 );
 make_type!(
     UInt16Type,
     u16,
     DataType::UInt16,
-    "An unsigned 16-bit integer type."
+    "Unsigned 16-bit integer type."
 );
 make_type!(
     UInt32Type,
     u32,
     DataType::UInt32,
-    "An unsigned 32-bit integer type."
+    "Unsigned 32-bit integer type."
 );
 make_type!(
     UInt64Type,
     u64,
     DataType::UInt64,
-    "An unsigned 64-bit integer type."
+    "Unsigned 64-bit integer type."
 );
 make_type!(
     Float16Type,
     f16,
     DataType::Float16,
-    "A 16-bit floating point number type."
+    "16-bit floating point number type."
 );
 make_type!(
     Float32Type,
     f32,
     DataType::Float32,
-    "A 32-bit floating point number type."
+    "32-bit floating point number type."
 );
 make_type!(
     Float64Type,
     f64,
     DataType::Float64,
-    "A 64-bit floating point number type."
+    "64-bit floating point number type."
 );
 make_type!(
     TimestampSecondType,
     i64,
     DataType::Timestamp(TimeUnit::Second, None),
-    "A timestamp second type with an optional timezone."
+    "Timestamp second type with an optional timezone."
 );
 make_type!(
     TimestampMillisecondType,
     i64,
     DataType::Timestamp(TimeUnit::Millisecond, None),
-    "A timestamp millisecond type with an optional timezone."
+    "Timestamp millisecond type with an optional timezone."
 );
 make_type!(
     TimestampMicrosecondType,
     i64,
     DataType::Timestamp(TimeUnit::Microsecond, None),
-    "A timestamp microsecond type with an optional timezone."
+    "Timestamp microsecond type with an optional timezone."
 );
 make_type!(
     TimestampNanosecondType,
     i64,
     DataType::Timestamp(TimeUnit::Nanosecond, None),
-    "A timestamp nanosecond type with an optional timezone."
+    "Timestamp nanosecond type with an optional timezone."
 );
 make_type!(
     Date32Type,
     i32,
     DataType::Date32,
-    "A 32-bit date type representing the elapsed time since UNIX epoch in days(32 bits)."
+    "32-bit date type: the elapsed time since UNIX epoch in days (32 bits)."
 );
 make_type!(
     Date64Type,
     i64,
     DataType::Date64,
-    "A 64-bit date type representing the elapsed time since UNIX epoch in milliseconds(64 bits)."
+    "64-bit date type: the elapsed time since UNIX epoch in milliseconds (64 bits). \
+    Values must be divisible by `86_400_000`. \
+    See [`DataType::Date64`] for more details."
 );
 make_type!(
     Time32SecondType,
     i32,
     DataType::Time32(TimeUnit::Second),
-    "A 32-bit time type representing the elapsed time since midnight in seconds."
+    "32-bit time type: the elapsed time since midnight in seconds."
 );
 make_type!(
     Time32MillisecondType,
     i32,
     DataType::Time32(TimeUnit::Millisecond),
-    "A 32-bit time type representing the elapsed time since midnight in milliseconds."
+    "32-bit time type: the elapsed time since midnight in milliseconds."
 );
 make_type!(
     Time64MicrosecondType,
     i64,
     DataType::Time64(TimeUnit::Microsecond),
-    "A 64-bit time type representing the elapsed time since midnight in microseconds."
+    "64-bit time type: the elapsed time since midnight in microseconds."
 );
 make_type!(
     Time64NanosecondType,
     i64,
     DataType::Time64(TimeUnit::Nanosecond),
-    "A 64-bit time type representing the elapsed time since midnight in nanoseconds."
+    "64-bit time type: the elapsed time since midnight in nanoseconds."
 );
 make_type!(
     IntervalYearMonthType,
     i32,
     DataType::Interval(IntervalUnit::YearMonth),
-    "A 32-bit “calendar” interval type representing the number of whole months."
+    "32-bit “calendar” interval type: the number of whole months."
 );
 make_type!(
     IntervalDayTimeType,
     IntervalDayTime,
     DataType::Interval(IntervalUnit::DayTime),
-    "A “calendar” interval type representing days and milliseconds. See [`IntervalDayTime`] for more details."
+    "“Calendar” interval type: days and milliseconds. See [`IntervalDayTime`] for more details."
 );
 make_type!(
     IntervalMonthDayNanoType,
     IntervalMonthDayNano,
     DataType::Interval(IntervalUnit::MonthDayNano),
-    r"A “calendar” interval type representing months, days, and nanoseconds. See [`IntervalMonthDayNano`] for more details."
+    r"“Calendar” interval type: months, days, and nanoseconds. See [`IntervalMonthDayNano`] for more details."
 );
 make_type!(
     DurationSecondType,
     i64,
     DataType::Duration(TimeUnit::Second),
-    "An elapsed time type in seconds."
+    "Elapsed time type: seconds."
 );
 make_type!(
     DurationMillisecondType,
     i64,
     DataType::Duration(TimeUnit::Millisecond),
-    "An elapsed time type in milliseconds."
+    "Elapsed time type: milliseconds."
 );
 make_type!(
     DurationMicrosecondType,
     i64,
     DataType::Duration(TimeUnit::Microsecond),
-    "An elapsed time type in microseconds."
+    "Elapsed time type: microseconds."
 );
 make_type!(
     DurationNanosecondType,
     i64,
     DataType::Duration(TimeUnit::Nanosecond),
-    "An elapsed time type in nanoseconds."
+    "Elapsed time type: nanoseconds."
 );
 
 /// A subtype of primitive type that represents legal dictionary keys.
diff --git a/arrow-flight/examples/flight_sql_server.rs b/arrow-flight/examples/flight_sql_server.rs
index d5168debc..81afecf85 100644
--- a/arrow-flight/examples/flight_sql_server.rs
+++ b/arrow-flight/examples/flight_sql_server.rs
@@ -46,9 +46,9 @@ use arrow_flight::sql::{
     ActionEndTransactionRequest, Any, CommandGetCatalogs, CommandGetCrossReference,
     CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys, CommandGetPrimaryKeys,
     CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables, CommandGetXdbcTypeInfo,
-    CommandPreparedStatementQuery, CommandPreparedStatementUpdate, CommandStatementQuery,
-    CommandStatementSubstraitPlan, CommandStatementUpdate, Nullable, ProstMessageExt, Searchable,
-    SqlInfo, TicketStatementQuery, XdbcDataType,
+    CommandPreparedStatementQuery, CommandPreparedStatementUpdate, CommandStatementIngest,
+    CommandStatementQuery, CommandStatementSubstraitPlan, CommandStatementUpdate, Nullable,
+    ProstMessageExt, Searchable, SqlInfo, TicketStatementQuery, XdbcDataType,
 };
 use arrow_flight::utils::batches_to_flight_data;
 use arrow_flight::{
@@ -615,6 +615,14 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(FAKE_UPDATE_RESULT)
     }
 
+    async fn do_put_statement_ingest(
+        &self,
+        _ticket: CommandStatementIngest,
+        _request: Request<PeekableFlightDataStream>,
+    ) -> Result<i64, Status> {
+        Ok(FAKE_UPDATE_RESULT)
+    }
+
     async fn do_put_substrait_plan(
         &self,
         _ticket: CommandStatementSubstraitPlan,
diff --git a/arrow-flight/src/client.rs b/arrow-flight/src/client.rs
index 3f62b256d..af3c8fba3 100644
--- a/arrow-flight/src/client.rs
+++ b/arrow-flight/src/client.rs
@@ -679,7 +679,7 @@ impl FlightClient {
 /// it encounters an error it uses the oneshot sender to
 /// notify the error and stop any further streaming. See `do_put` or
 /// `do_exchange` for it's uses.
-struct FallibleRequestStream<T, E> {
+pub(crate) struct FallibleRequestStream<T, E> {
     /// sender to notify error
     sender: Option<Sender<E>>,
     /// fallible stream
@@ -687,7 +687,7 @@ struct FallibleRequestStream<T, E> {
 }
 
 impl<T, E> FallibleRequestStream<T, E> {
-    fn new(
+    pub(crate) fn new(
         sender: Sender<E>,
         fallible_stream: Pin<Box<dyn Stream<Item = std::result::Result<T, E>> + Send + 'static>>,
     ) -> Self {
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index 91790898b..9f9963c92 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -24,6 +24,7 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use tonic::metadata::AsciiMetadataKey;
 
+use crate::client::FallibleRequestStream;
 use crate::decode::FlightRecordBatchStream;
 use crate::encode::FlightDataEncoderBuilder;
 use crate::error::FlightError;
@@ -39,8 +40,8 @@ use crate::sql::{
     CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys,
     CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables,
     CommandGetXdbcTypeInfo, CommandPreparedStatementQuery, CommandPreparedStatementUpdate,
-    CommandStatementQuery, CommandStatementUpdate, DoPutPreparedStatementResult, DoPutUpdateResult,
-    ProstMessageExt, SqlInfo,
+    CommandStatementIngest, CommandStatementQuery, CommandStatementUpdate,
+    DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt, SqlInfo,
 };
 use crate::trailers::extract_lazy_trailers;
 use crate::{
@@ -53,10 +54,10 @@ use arrow_ipc::convert::fb_to_schema;
 use arrow_ipc::reader::read_record_batch;
 use arrow_ipc::{root_as_message, MessageHeader};
 use arrow_schema::{ArrowError, Schema, SchemaRef};
-use futures::{stream, TryStreamExt};
+use futures::{stream, Stream, TryStreamExt};
 use prost::Message;
 use tonic::transport::Channel;
-use tonic::{IntoRequest, Streaming};
+use tonic::{IntoRequest, IntoStreamingRequest, Streaming};
 
 /// A FlightSQLServiceClient is an endpoint for retrieving or storing Arrow data
 /// by FlightSQL protocol.
@@ -227,6 +228,52 @@ impl FlightSqlServiceClient<Channel> {
         Ok(result.record_count)
     }
 
+    /// Execute a bulk ingest on the server and return the number of records added
+    pub async fn execute_ingest<S>(
+        &mut self,
+        command: CommandStatementIngest,
+        stream: S,
+    ) -> Result<i64, ArrowError>
+    where
+        S: Stream<Item = crate::error::Result<RecordBatch>> + Send + 'static,
+    {
+        let (sender, receiver) = futures::channel::oneshot::channel();
+
+        let descriptor = FlightDescriptor::new_cmd(command.as_any().encode_to_vec());
+        let flight_data = FlightDataEncoderBuilder::new()
+            .with_flight_descriptor(Some(descriptor))
+            .build(stream);
+
+        // Intercept client errors and send them to the one shot channel above
+        let flight_data = Box::pin(flight_data);
+        let flight_data: FallibleRequestStream<FlightData, FlightError> =
+            FallibleRequestStream::new(sender, flight_data);
+
+        let req = self.set_request_headers(flight_data.into_streaming_request())?;
+        let mut result = self
+            .flight_client
+            .do_put(req)
+            .await
+            .map_err(status_to_arrow_error)?
+            .into_inner();
+
+        // check if the there were any errors in the input stream provided note
+        // if receiver.await fails, it means the sender was dropped and there is
+        // no message to return.
+        if let Ok(msg) = receiver.await {
+            return Err(ArrowError::ExternalError(Box::new(msg)));
+        }
+
+        let result = result
+            .message()
+            .await
+            .map_err(status_to_arrow_error)?
+            .unwrap();
+        let any = Any::decode(&*result.app_metadata).map_err(decode_error_to_arrow_error)?;
+        let result: DoPutUpdateResult = any.unpack()?.unwrap();
+        Ok(result.record_count)
+    }
+
     /// Request a list of catalogs as tabular FlightInfo results
     pub async fn get_catalogs(&mut self) -> Result<FlightInfo, ArrowError> {
         self.get_flight_info_for_command(CommandGetCatalogs {})
diff --git a/arrow-flight/src/sql/mod.rs b/arrow-flight/src/sql/mod.rs
index 61eb67b69..453f608d3 100644
--- a/arrow-flight/src/sql/mod.rs
+++ b/arrow-flight/src/sql/mod.rs
@@ -50,6 +50,10 @@ mod gen {
 }
 
 pub use gen::action_end_transaction_request::EndTransaction;
+pub use gen::command_statement_ingest::table_definition_options::{
+    TableExistsOption, TableNotExistOption,
+};
+pub use gen::command_statement_ingest::TableDefinitionOptions;
 pub use gen::ActionBeginSavepointRequest;
 pub use gen::ActionBeginSavepointResult;
 pub use gen::ActionBeginTransactionRequest;
@@ -74,6 +78,7 @@ pub use gen::CommandGetTables;
 pub use gen::CommandGetXdbcTypeInfo;
 pub use gen::CommandPreparedStatementQuery;
 pub use gen::CommandPreparedStatementUpdate;
+pub use gen::CommandStatementIngest;
 pub use gen::CommandStatementQuery;
 pub use gen::CommandStatementSubstraitPlan;
 pub use gen::CommandStatementUpdate;
@@ -250,11 +255,12 @@ prost_message_ext!(
     CommandGetXdbcTypeInfo,
     CommandPreparedStatementQuery,
     CommandPreparedStatementUpdate,
+    CommandStatementIngest,
     CommandStatementQuery,
     CommandStatementSubstraitPlan,
     CommandStatementUpdate,
-    DoPutUpdateResult,
     DoPutPreparedStatementResult,
+    DoPutUpdateResult,
     TicketStatementQuery,
 );
 
diff --git a/arrow-flight/src/sql/server.rs b/arrow-flight/src/sql/server.rs
index b47691c7d..e348367a9 100644
--- a/arrow-flight/src/sql/server.rs
+++ b/arrow-flight/src/sql/server.rs
@@ -32,9 +32,9 @@ use super::{
     CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys,
     CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables,
     CommandGetXdbcTypeInfo, CommandPreparedStatementQuery, CommandPreparedStatementUpdate,
-    CommandStatementQuery, CommandStatementSubstraitPlan, CommandStatementUpdate,
-    DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt, SqlInfo,
-    TicketStatementQuery,
+    CommandStatementIngest, CommandStatementQuery, CommandStatementSubstraitPlan,
+    CommandStatementUpdate, DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt,
+    SqlInfo, TicketStatementQuery,
 };
 use crate::{
     flight_service_server::FlightService, gen::PollInfo, Action, ActionType, Criteria, Empty,
@@ -397,6 +397,17 @@ pub trait FlightSqlService: Sync + Send + Sized + 'static {
         ))
     }
 
+    /// Execute a bulk ingestion.
+    async fn do_put_statement_ingest(
+        &self,
+        _ticket: CommandStatementIngest,
+        _request: Request<PeekableFlightDataStream>,
+    ) -> Result<i64, Status> {
+        Err(Status::unimplemented(
+            "do_put_statement_ingest has no default implementation",
+        ))
+    }
+
     /// Bind parameters to given prepared statement.
     ///
     /// Returns an opaque handle that the client should pass
@@ -713,6 +724,14 @@ where
                 })]);
                 Ok(Response::new(Box::pin(output)))
             }
+            Command::CommandStatementIngest(command) => {
+                let record_count = self.do_put_statement_ingest(command, request).await?;
+                let result = DoPutUpdateResult { record_count };
+                let output = futures::stream::iter(vec![Ok(PutResult {
+                    app_metadata: result.as_any().encode_to_vec().into(),
+                })]);
+                Ok(Response::new(Box::pin(output)))
+            }
             Command::CommandPreparedStatementQuery(command) => {
                 let result = self
                     .do_put_prepared_statement_query(command, request)
diff --git a/arrow-schema/src/datatype.rs b/arrow-schema/src/datatype.rs
index 38e50a17a..1848c8b3f 100644
--- a/arrow-schema/src/datatype.rs
+++ b/arrow-schema/src/datatype.rs
@@ -203,22 +203,34 @@ pub enum DataType {
     /// A signed 64-bit date representing the elapsed time since UNIX epoch (1970-01-01)
     /// in milliseconds.
     ///
-    /// According to the specification (see [Schema.fbs]), this should be treated as the number of
-    /// days, in milliseconds, since the UNIX epoch. Therefore, values must be evenly divisible by
-    /// `86_400_000` (the number of milliseconds in a standard day).
+    /// # Valid Ranges
     ///
-    /// The reason for this is for compatibility with other language's native libraries,
-    /// such as Java, which historically lacked a dedicated date type
-    /// and only supported timestamps.
+    /// According to the Arrow specification ([Schema.fbs]), values of Date64
+    /// are treated as the number of *days*, in milliseconds, since the UNIX
+    /// epoch. Therefore, values of this type  must be evenly divisible by
+    /// `86_400_000`, the number of milliseconds in a standard day.
     ///
-    /// Practically, validation that values of this type are evenly divisible by `86_400_000` is not enforced
-    /// by this library for performance and usability reasons. Date64 values will be treated similarly to the
-    /// `Timestamp(TimeUnit::Millisecond, None)` type, in that its values will be printed showing the time of
-    /// day if the value does not represent an exact day, and arithmetic can be done at the millisecond
-    /// granularity to change the time represented.
+    /// It is not valid to store milliseconds that do not represent an exact
+    /// day. The reason for this restriction is compatibility with other
+    /// language's native libraries (specifically Java), which historically
+    /// lacked a dedicated date type and only supported timestamps.
     ///
-    /// Users should prefer using Date32 to cleanly represent the number of days, or one of the Timestamp
-    /// variants to include time as part of the representation, depending on their use case.
+    /// # Validation
+    ///
+    /// This library does not validate or enforce that Date64 values are evenly
+    /// divisible by `86_400_000`  for performance and usability reasons. Date64
+    /// values are treated similarly to `Timestamp(TimeUnit::Millisecond,
+    /// None)`: values will be displayed with a time of day if the value does
+    /// not represent an exact day, and arithmetic will be done at the
+    /// millisecond granularity.
+    ///
+    /// # Recommendation
+    ///
+    /// Users should prefer [`DataType::Date32`] to cleanly represent the number
+    /// of days, or one of the Timestamp variants to include time as part of the
+    /// representation, depending on their use case.
+    ///
+    /// # Further Reading
     ///
     /// For more details, see [#5288](https://github.com/apache/arrow-rs/issues/5288).
     ///
diff --git a/object_store/CHANGELOG.md b/object_store/CHANGELOG.md
index dc7117112..18dde1119 100644
--- a/object_store/CHANGELOG.md
+++ b/object_store/CHANGELOG.md
@@ -31,6 +31,7 @@
 
 **Merged pull requests:**
 
+- Add LICENSE and NOTICE files to object_store  [\#6234](https://github.com/apache/arrow-rs/pull/6234) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
 - feat\(object\_store\):  add `PermissionDenied` variant to top-level error [\#6194](https://github.com/apache/arrow-rs/pull/6194) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([kyle-mccarthy](https://github.com/kyle-mccarthy))
 - Update object store MSRV to `1.64` [\#6123](https://github.com/apache/arrow-rs/pull/6123) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
 - Fix clippy in object\_store crate [\#6120](https://github.com/apache/arrow-rs/pull/6120) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
diff --git a/object_store/CONTRIBUTING.md b/object_store/CONTRIBUTING.md
index 4b0ef1f0e..5444ec743 100644
--- a/object_store/CONTRIBUTING.md
+++ b/object_store/CONTRIBUTING.md
@@ -101,6 +101,54 @@ export AWS_SERVER_SIDE_ENCRYPTION=aws:kms:dsse
 cargo test --features aws
 ```
 
+#### SSE-C Encryption tests
+
+Unfortunately, localstack does not support SSE-C encryption (https://github.com/localstack/localstack/issues/11356).
+
+We will use [MinIO](https://min.io/docs/minio/container/operations/server-side-encryption.html) to test SSE-C encryption.
+
+First, create a self-signed certificate to enable HTTPS for MinIO, as SSE-C requires HTTPS.
+
+```shell
+mkdir ~/certs
+cd ~/certs
+openssl genpkey -algorithm RSA -out private.key
+openssl req -new -key private.key -out request.csr -subj "/C=US/ST=State/L=City/O=Organization/OU=Unit/CN=example.com/emailAddress=email@example.com"
+openssl x509 -req -days 365 -in request.csr -signkey private.key -out public.crt
+rm request.csr
+```
+
+Second, start MinIO with the self-signed certificate.
+
+```shell
+docker run -d \
+  -p 9000:9000 \
+  --name minio \
+  -v ${HOME}/certs:/root/.minio/certs \
+  -e "MINIO_ROOT_USER=minio" \
+  -e "MINIO_ROOT_PASSWORD=minio123" \
+  minio/minio server /data
+```
+
+Create a test bucket.
+
+```shell
+export AWS_BUCKET_NAME=test-bucket
+export AWS_ACCESS_KEY_ID=minio
+export AWS_SECRET_ACCESS_KEY=minio123
+export AWS_ENDPOINT=https://localhost:9000
+aws s3 mb s3://test-bucket --endpoint-url=https://localhost:9000 --no-verify-ssl
+```
+
+Run the tests. The real test is `test_s3_ssec_encryption_with_minio()`
+
+```shell
+export TEST_S3_SSEC_ENCRYPTION=1
+cargo test --features aws --package object_store --lib aws::tests::test_s3_ssec_encryption_with_minio -- --exact --nocapture
+```
+
+
+
 ### Azure
 
 To test the Azure integration
diff --git a/object_store/LICENSE.txt b/object_store/LICENSE.txt
new file mode 100644
index 000000000..de4b130f3
--- /dev/null
+++ b/object_store/LICENSE.txt
@@ -0,0 +1,204 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed 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.
+
+
diff --git a/object_store/NOTICE.txt b/object_store/NOTICE.txt
new file mode 100644
index 000000000..0a23eee66
--- /dev/null
+++ b/object_store/NOTICE.txt
@@ -0,0 +1,5 @@
+Apache Arrow Object Store
+Copyright 2020-2024 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index ffef3fb33..574345c38 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -26,7 +26,10 @@ use crate::aws::{
 use crate::client::TokenCredentialProvider;
 use crate::config::ConfigValue;
 use crate::{ClientConfigKey, ClientOptions, Result, RetryConfig, StaticCredentialProvider};
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use itertools::Itertools;
+use md5::{Digest, Md5};
 use reqwest::header::{HeaderMap, HeaderValue};
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
@@ -73,7 +76,7 @@ enum Error {
     #[snafu(display("Invalid Zone suffix for bucket '{bucket}'"))]
     ZoneSuffix { bucket: String },
 
-    #[snafu(display("Invalid encryption type: {}. Valid values are \"AES256\", \"sse:kms\", and \"sse:kms:dsse\".", passed))]
+    #[snafu(display("Invalid encryption type: {}. Valid values are \"AES256\", \"sse:kms\", \"sse:kms:dsse\" and \"sse-c\".", passed))]
     InvalidEncryptionType { passed: String },
 
     #[snafu(display(
@@ -166,6 +169,8 @@ pub struct AmazonS3Builder {
     encryption_type: Option<ConfigValue<S3EncryptionType>>,
     encryption_kms_key_id: Option<String>,
     encryption_bucket_key_enabled: Option<ConfigValue<bool>>,
+    /// base64-encoded 256-bit customer encryption key for SSE-C.
+    encryption_customer_key_base64: Option<String>,
 }
 
 /// Configuration keys for [`AmazonS3Builder`]
@@ -394,6 +399,9 @@ impl FromStr for AmazonS3ConfigKey {
             "aws_sse_bucket_key_enabled" => {
                 Ok(Self::Encryption(S3EncryptionConfigKey::BucketKeyEnabled))
             }
+            "aws_sse_customer_key_base64" => Ok(Self::Encryption(
+                S3EncryptionConfigKey::CustomerEncryptionKey,
+            )),
             _ => match s.parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
@@ -511,6 +519,9 @@ impl AmazonS3Builder {
                 S3EncryptionConfigKey::BucketKeyEnabled => {
                     self.encryption_bucket_key_enabled = Some(ConfigValue::Deferred(value.into()))
                 }
+                S3EncryptionConfigKey::CustomerEncryptionKey => {
+                    self.encryption_customer_key_base64 = Some(value.into())
+                }
             },
         };
         self
@@ -566,6 +577,9 @@ impl AmazonS3Builder {
                     .encryption_bucket_key_enabled
                     .as_ref()
                     .map(ToString::to_string),
+                S3EncryptionConfigKey::CustomerEncryptionKey => {
+                    self.encryption_customer_key_base64.clone()
+                }
             },
         }
     }
@@ -813,6 +827,14 @@ impl AmazonS3Builder {
         self
     }
 
+    /// Use SSE-C for server side encryption.
+    /// Must pass the *base64-encoded* 256-bit customer encryption key.
+    pub fn with_ssec_encryption(mut self, customer_key_base64: impl Into<String>) -> Self {
+        self.encryption_type = Some(ConfigValue::Parsed(S3EncryptionType::SseC));
+        self.encryption_customer_key_base64 = customer_key_base64.into().into();
+        self
+    }
+
     /// Set whether to enable bucket key for server side encryption. This overrides
     /// the bucket default setting for bucket keys.
     ///
@@ -953,6 +975,7 @@ impl AmazonS3Builder {
                 self.encryption_bucket_key_enabled
                     .map(|val| val.get())
                     .transpose()?,
+                self.encryption_customer_key_base64,
             )?
         } else {
             S3EncryptionHeaders::default()
@@ -994,15 +1017,14 @@ fn parse_bucket_az(bucket: &str) -> Option<&str> {
 /// These options are used to configure server-side encryption for S3 objects.
 /// To configure them, pass them to [`AmazonS3Builder::with_config`].
 ///
-/// Both [SSE-KMS] and [DSSE-KMS] are supported. [SSE-C] is not yet supported.
-///
+/// [SSE-S3]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html
 /// [SSE-KMS]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingKMSEncryption.html
 /// [DSSE-KMS]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingDSSEncryption.html
 /// [SSE-C]: https://docs.aws.amazon.com/AmazonS3/latest/userguide/ServerSideEncryptionCustomerKeys.html
 #[derive(PartialEq, Eq, Hash, Clone, Debug, Copy, Serialize, Deserialize)]
 #[non_exhaustive]
 pub enum S3EncryptionConfigKey {
-    /// Type of encryption to use. If set, must be one of "AES256", "aws:kms", or "aws:kms:dsse".
+    /// Type of encryption to use. If set, must be one of "AES256" (SSE-S3), "aws:kms" (SSE-KMS), "aws:kms:dsse" (DSSE-KMS) or "sse-c".
     ServerSideEncryption,
     /// The KMS key ID to use for server-side encryption. If set, ServerSideEncryption
     /// must be "aws:kms" or "aws:kms:dsse".
@@ -1010,6 +1032,10 @@ pub enum S3EncryptionConfigKey {
     /// If set to true, will use the bucket's default KMS key for server-side encryption.
     /// If set to false, will disable the use of the bucket's default KMS key for server-side encryption.
     BucketKeyEnabled,
+
+    /// The base64 encoded, 256-bit customer encryption key to use for server-side encryption.
+    /// If set, ServerSideEncryption must be "sse-c".
+    CustomerEncryptionKey,
 }
 
 impl AsRef<str> for S3EncryptionConfigKey {
@@ -1018,6 +1044,7 @@ impl AsRef<str> for S3EncryptionConfigKey {
             Self::ServerSideEncryption => "aws_server_side_encryption",
             Self::KmsKeyId => "aws_sse_kms_key_id",
             Self::BucketKeyEnabled => "aws_sse_bucket_key_enabled",
+            Self::CustomerEncryptionKey => "aws_sse_customer_key_base64",
         }
     }
 }
@@ -1027,6 +1054,7 @@ enum S3EncryptionType {
     S3,
     SseKms,
     DsseKms,
+    SseC,
 }
 
 impl crate::config::Parse for S3EncryptionType {
@@ -1035,6 +1063,7 @@ impl crate::config::Parse for S3EncryptionType {
             "AES256" => Ok(Self::S3),
             "aws:kms" => Ok(Self::SseKms),
             "aws:kms:dsse" => Ok(Self::DsseKms),
+            "sse-c" => Ok(Self::SseC),
             _ => Err(Error::InvalidEncryptionType { passed: s.into() }.into()),
         }
     }
@@ -1046,6 +1075,7 @@ impl From<&S3EncryptionType> for &'static str {
             S3EncryptionType::S3 => "AES256",
             S3EncryptionType::SseKms => "aws:kms",
             S3EncryptionType::DsseKms => "aws:kms:dsse",
+            S3EncryptionType::SseC => "sse-c",
         }
     }
 }
@@ -1062,37 +1092,87 @@ impl std::fmt::Display for S3EncryptionType {
 /// Whether these headers are sent depends on both the kind of encryption set
 /// and the kind of request being made.
 #[derive(Default, Clone, Debug)]
-pub struct S3EncryptionHeaders(HeaderMap);
+pub(super) struct S3EncryptionHeaders(pub HeaderMap);
 
 impl S3EncryptionHeaders {
     fn try_new(
         encryption_type: &S3EncryptionType,
-        key_id: Option<String>,
+        encryption_kms_key_id: Option<String>,
         bucket_key_enabled: Option<bool>,
+        encryption_customer_key_base64: Option<String>,
     ) -> Result<Self> {
         let mut headers = HeaderMap::new();
-        // Note: if we later add support for SSE-C, we should be sure to use
-        // HeaderValue::set_sensitive to prevent the key from being logged.
-        headers.insert(
-            "x-amz-server-side-encryption",
-            HeaderValue::from_static(encryption_type.into()),
-        );
-        if let Some(key_id) = key_id {
-            headers.insert(
-                "x-amz-server-side-encryption-aws-kms-key-id",
-                key_id
-                    .try_into()
-                    .map_err(|err| Error::InvalidEncryptionHeader {
-                        header: "kms-key-id",
-                        source: Box::new(err),
-                    })?,
-            );
-        }
-        if let Some(bucket_key_enabled) = bucket_key_enabled {
-            headers.insert(
-                "x-amz-server-side-encryption-bucket-key-enabled",
-                HeaderValue::from_static(if bucket_key_enabled { "true" } else { "false" }),
-            );
+        match encryption_type {
+            S3EncryptionType::S3 | S3EncryptionType::SseKms | S3EncryptionType::DsseKms => {
+                headers.insert(
+                    "x-amz-server-side-encryption",
+                    HeaderValue::from_static(encryption_type.into()),
+                );
+                if let Some(key_id) = encryption_kms_key_id {
+                    headers.insert(
+                        "x-amz-server-side-encryption-aws-kms-key-id",
+                        key_id
+                            .try_into()
+                            .map_err(|err| Error::InvalidEncryptionHeader {
+                                header: "kms-key-id",
+                                source: Box::new(err),
+                            })?,
+                    );
+                }
+                if let Some(bucket_key_enabled) = bucket_key_enabled {
+                    headers.insert(
+                        "x-amz-server-side-encryption-bucket-key-enabled",
+                        HeaderValue::from_static(if bucket_key_enabled { "true" } else { "false" }),
+                    );
+                }
+            }
+            S3EncryptionType::SseC => {
+                headers.insert(
+                    "x-amz-server-side-encryption-customer-algorithm",
+                    HeaderValue::from_static("AES256"),
+                );
+                if let Some(key) = encryption_customer_key_base64 {
+                    let mut header_value: HeaderValue =
+                        key.clone()
+                            .try_into()
+                            .map_err(|err| Error::InvalidEncryptionHeader {
+                                header: "x-amz-server-side-encryption-customer-key",
+                                source: Box::new(err),
+                            })?;
+                    header_value.set_sensitive(true);
+                    headers.insert("x-amz-server-side-encryption-customer-key", header_value);
+
+                    let decoded_key = BASE64_STANDARD.decode(key.as_bytes()).map_err(|err| {
+                        Error::InvalidEncryptionHeader {
+                            header: "x-amz-server-side-encryption-customer-key",
+                            source: Box::new(err),
+                        }
+                    })?;
+                    let mut hasher = Md5::new();
+                    hasher.update(decoded_key);
+                    let md5 = BASE64_STANDARD.encode(hasher.finalize());
+                    let mut md5_header_value: HeaderValue =
+                        md5.try_into()
+                            .map_err(|err| Error::InvalidEncryptionHeader {
+                                header: "x-amz-server-side-encryption-customer-key-MD5",
+                                source: Box::new(err),
+                            })?;
+                    md5_header_value.set_sensitive(true);
+                    headers.insert(
+                        "x-amz-server-side-encryption-customer-key-MD5",
+                        md5_header_value,
+                    );
+                } else {
+                    return Err(Error::InvalidEncryptionHeader {
+                        header: "x-amz-server-side-encryption-customer-key",
+                        source: Box::new(std::io::Error::new(
+                            std::io::ErrorKind::InvalidInput,
+                            "Missing customer key",
+                        )),
+                    }
+                    .into());
+                }
+            }
         }
         Ok(Self(headers))
     }
@@ -1162,7 +1242,11 @@ mod tests {
             .with_config(AmazonS3ConfigKey::UnsignedPayload, "true")
             .with_config("aws_server_side_encryption".parse().unwrap(), "AES256")
             .with_config("aws_sse_kms_key_id".parse().unwrap(), "some_key_id")
-            .with_config("aws_sse_bucket_key_enabled".parse().unwrap(), "true");
+            .with_config("aws_sse_bucket_key_enabled".parse().unwrap(), "true")
+            .with_config(
+                "aws_sse_customer_key_base64".parse().unwrap(),
+                "some_customer_key",
+            );
 
         assert_eq!(
             builder
@@ -1216,6 +1300,12 @@ mod tests {
                 .unwrap(),
             "true"
         );
+        assert_eq!(
+            builder
+                .get_config_value(&"aws_sse_customer_key_base64".parse().unwrap())
+                .unwrap(),
+            "some_customer_key"
+        );
     }
 
     #[test]
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index ab4da86f5..007e27108 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -181,7 +181,7 @@ pub struct S3Config {
     pub checksum: Option<Checksum>,
     pub copy_if_not_exists: Option<S3CopyIfNotExists>,
     pub conditional_put: Option<S3ConditionalPut>,
-    pub encryption_headers: S3EncryptionHeaders,
+    pub(super) encryption_headers: S3EncryptionHeaders,
 }
 
 impl S3Config {
@@ -522,10 +522,47 @@ impl S3Client {
     /// Make an S3 Copy request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html>
     pub fn copy_request<'a>(&'a self, from: &Path, to: &'a Path) -> Request<'a> {
         let source = format!("{}/{}", self.config.bucket, encode_path(from));
+
+        let mut copy_source_encryption_headers = HeaderMap::new();
+        if let Some(customer_algorithm) = self
+            .config
+            .encryption_headers
+            .0
+            .get("x-amz-server-side-encryption-customer-algorithm")
+        {
+            copy_source_encryption_headers.insert(
+                "x-amz-copy-source-server-side-encryption-customer-algorithm",
+                customer_algorithm.clone(),
+            );
+        }
+        if let Some(customer_key) = self
+            .config
+            .encryption_headers
+            .0
+            .get("x-amz-server-side-encryption-customer-key")
+        {
+            copy_source_encryption_headers.insert(
+                "x-amz-copy-source-server-side-encryption-customer-key",
+                customer_key.clone(),
+            );
+        }
+        if let Some(customer_key_md5) = self
+            .config
+            .encryption_headers
+            .0
+            .get("x-amz-server-side-encryption-customer-key-MD5")
+        {
+            copy_source_encryption_headers.insert(
+                "x-amz-copy-source-server-side-encryption-customer-key-MD5",
+                customer_key_md5.clone(),
+            );
+        }
+
         self.request(Method::PUT, to)
             .idempotent(true)
             .header(&COPY_SOURCE_HEADER, &source)
             .headers(self.config.encryption_headers.clone().into())
+            .headers(copy_source_encryption_headers)
             .with_session_creds(false)
     }
 
@@ -562,13 +599,21 @@ impl S3Client {
     ) -> Result<PartId> {
         let part = (part_idx + 1).to_string();
 
-        let response = self
+        let mut request = self
             .request(Method::PUT, path)
             .with_payload(data)
             .query(&[("partNumber", &part), ("uploadId", upload_id)])
-            .idempotent(true)
-            .send()
-            .await?;
+            .idempotent(true);
+        if self
+            .config
+            .encryption_headers
+            .0
+            .contains_key("x-amz-server-side-encryption-customer-algorithm")
+        {
+            // If SSE-C is used, we must include the encryption headers in every upload request.
+            request = request.with_encryption_headers();
+        }
+        let response = request.send().await?;
 
         let content_id = get_etag(response.headers()).context(MetadataSnafu)?;
         Ok(PartId { content_id })
@@ -660,6 +705,7 @@ impl GetClient for S3Client {
         };
 
         let mut builder = self.client.request(method, url);
+        builder = builder.headers(self.config.encryption_headers.clone().into());
 
         if let Some(v) = &options.version {
             builder = builder.query(&[("versionId", v)])
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index f5204a536..4a773e7a1 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -60,7 +60,7 @@ mod dynamo;
 mod precondition;
 mod resolve;
 
-pub use builder::{AmazonS3Builder, AmazonS3ConfigKey, S3EncryptionHeaders};
+pub use builder::{AmazonS3Builder, AmazonS3ConfigKey};
 pub use checksum::Checksum;
 pub use dynamo::DynamoCommit;
 pub use precondition::{S3ConditionalPut, S3CopyIfNotExists};
@@ -412,6 +412,9 @@ mod tests {
     use crate::client::get::GetClient;
     use crate::integration::*;
     use crate::tests::*;
+    use crate::ClientOptions;
+    use base64::prelude::BASE64_STANDARD;
+    use base64::Engine;
     use hyper::HeaderMap;
 
     const NON_EXISTENT_NAME: &str = "nonexistentname";
@@ -605,4 +608,67 @@ mod tests {
             store.delete(location).await.unwrap();
         }
     }
+
+    /// See CONTRIBUTING.md for the MinIO setup for this test.
+    #[tokio::test]
+    async fn test_s3_ssec_encryption_with_minio() {
+        if std::env::var("TEST_S3_SSEC_ENCRYPTION").is_err() {
+            eprintln!("Skipping S3 SSE-C encryption test");
+            return;
+        }
+        eprintln!("Running S3 SSE-C encryption test");
+
+        let customer_key = "1234567890abcdef1234567890abcdef";
+        let expected_md5 = "JMwgiexXqwuPqIPjYFmIZQ==";
+
+        let store = AmazonS3Builder::from_env()
+            .with_ssec_encryption(BASE64_STANDARD.encode(customer_key))
+            .with_client_options(ClientOptions::default().with_allow_invalid_certificates(true))
+            .build()
+            .unwrap();
+
+        let data = PutPayload::from(vec![3u8; 1024]);
+
+        let locations = [
+            Path::from("test-encryption-1"),
+            Path::from("test-encryption-2"),
+            Path::from("test-encryption-3"),
+        ];
+
+        // Test put with sse-c.
+        store.put(&locations[0], data.clone()).await.unwrap();
+
+        // Test copy with sse-c.
+        store.copy(&locations[0], &locations[1]).await.unwrap();
+
+        // Test multipart upload with sse-c.
+        let mut upload = store.put_multipart(&locations[2]).await.unwrap();
+        upload.put_part(data.clone()).await.unwrap();
+        upload.complete().await.unwrap();
+
+        // Test get with sse-c.
+        for location in &locations {
+            let res = store
+                .client
+                .get_request(location, GetOptions::default())
+                .await
+                .unwrap();
+            let headers = res.headers();
+            assert_eq!(
+                headers
+                    .get("x-amz-server-side-encryption-customer-algorithm")
+                    .expect("object is not encrypted with SSE-C"),
+                "AES256"
+            );
+
+            assert_eq!(
+                headers
+                    .get("x-amz-server-side-encryption-customer-key-MD5")
+                    .expect("object is not encrypted with SSE-C"),
+                expected_md5
+            );
+
+            store.delete(location).await.unwrap();
+        }
+    }
 }
diff --git a/parquet/benches/encoding.rs b/parquet/benches/encoding.rs
index bc18a49da..8e61666e6 100644
--- a/parquet/benches/encoding.rs
+++ b/parquet/benches/encoding.rs
@@ -17,7 +17,7 @@
 
 use criterion::*;
 use half::f16;
-use parquet::basic::Encoding;
+use parquet::basic::{Encoding, Type as ParquetType};
 use parquet::data_type::{
     DataType, DoubleType, FixedLenByteArray, FixedLenByteArrayType, FloatType,
 };
@@ -35,7 +35,10 @@ fn bench_typed<T: DataType>(
 ) {
     let name = format!(
         "dtype={}, encoding={:?}",
-        std::any::type_name::<T::T>(),
+        match T::get_physical_type() {
+            ParquetType::FIXED_LEN_BYTE_ARRAY => format!("FixedLenByteArray({type_length})"),
+            _ => std::any::type_name::<T::T>().to_string(),
+        },
         encoding
     );
     let column_desc_ptr = ColumnDescPtr::new(ColumnDescriptor::new(
diff --git a/parquet/src/arrow/array_reader/fixed_len_byte_array.rs b/parquet/src/arrow/array_reader/fixed_len_byte_array.rs
index 8098f3240..3b2600c54 100644
--- a/parquet/src/arrow/array_reader/fixed_len_byte_array.rs
+++ b/parquet/src/arrow/array_reader/fixed_len_byte_array.rs
@@ -27,7 +27,7 @@ use crate::column::reader::decoder::ColumnValueDecoder;
 use crate::errors::{ParquetError, Result};
 use crate::schema::types::ColumnDescPtr;
 use arrow_array::{
-    ArrayRef, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float16Array,
+    Array, ArrayRef, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float16Array,
     IntervalDayTimeArray, IntervalYearMonthArray,
 };
 use arrow_buffer::{i256, Buffer, IntervalDayTime};
@@ -165,57 +165,68 @@ impl ArrayReader for FixedLenByteArrayReader {
         // TODO: An improvement might be to do this conversion on read
         let array: ArrayRef = match &self.data_type {
             ArrowType::Decimal128(p, s) => {
-                let decimal = binary
-                    .iter()
-                    .map(|opt| Some(i128::from_be_bytes(sign_extend_be(opt?))))
-                    .collect::<Decimal128Array>()
+                // We can simply reuse the null buffer from `binary` rather than recomputing it
+                // (as was the case when we simply used `collect` to produce the new array).
+                // The same applies to the transformations below.
+                let nulls = binary.nulls().cloned();
+                let decimal = binary.iter().map(|o| match o {
+                    Some(b) => i128::from_be_bytes(sign_extend_be(b)),
+                    None => i128::default(),
+                });
+                let decimal = Decimal128Array::from_iter_values_with_nulls(decimal, nulls)
                     .with_precision_and_scale(*p, *s)?;
-
                 Arc::new(decimal)
             }
             ArrowType::Decimal256(p, s) => {
-                let decimal = binary
-                    .iter()
-                    .map(|opt| Some(i256::from_be_bytes(sign_extend_be(opt?))))
-                    .collect::<Decimal256Array>()
+                let nulls = binary.nulls().cloned();
+                let decimal = binary.iter().map(|o| match o {
+                    Some(b) => i256::from_be_bytes(sign_extend_be(b)),
+                    None => i256::default(),
+                });
+                let decimal = Decimal256Array::from_iter_values_with_nulls(decimal, nulls)
                     .with_precision_and_scale(*p, *s)?;
-
                 Arc::new(decimal)
             }
             ArrowType::Interval(unit) => {
+                let nulls = binary.nulls().cloned();
                 // An interval is stored as 3x 32-bit unsigned integers storing months, days,
                 // and milliseconds
                 match unit {
-                    IntervalUnit::YearMonth => Arc::new(
-                        binary
-                            .iter()
-                            .map(|o| o.map(|b| i32::from_le_bytes(b[0..4].try_into().unwrap())))
-                            .collect::<IntervalYearMonthArray>(),
-                    ) as ArrayRef,
-                    IntervalUnit::DayTime => Arc::new(
-                        binary
-                            .iter()
-                            .map(|o| {
-                                o.map(|b| {
-                                    IntervalDayTime::new(
-                                        i32::from_le_bytes(b[4..8].try_into().unwrap()),
-                                        i32::from_le_bytes(b[8..12].try_into().unwrap()),
-                                    )
-                                })
-                            })
-                            .collect::<IntervalDayTimeArray>(),
-                    ) as ArrayRef,
+                    IntervalUnit::YearMonth => {
+                        let iter = binary.iter().map(|o| match o {
+                            Some(b) => i32::from_le_bytes(b[0..4].try_into().unwrap()),
+                            None => i32::default(),
+                        });
+                        let interval =
+                            IntervalYearMonthArray::from_iter_values_with_nulls(iter, nulls);
+                        Arc::new(interval) as ArrayRef
+                    }
+                    IntervalUnit::DayTime => {
+                        let iter = binary.iter().map(|o| match o {
+                            Some(b) => IntervalDayTime::new(
+                                i32::from_le_bytes(b[4..8].try_into().unwrap()),
+                                i32::from_le_bytes(b[8..12].try_into().unwrap()),
+                            ),
+                            None => IntervalDayTime::default(),
+                        });
+                        let interval =
+                            IntervalDayTimeArray::from_iter_values_with_nulls(iter, nulls);
+                        Arc::new(interval) as ArrayRef
+                    }
                     IntervalUnit::MonthDayNano => {
                         return Err(nyi_err!("MonthDayNano intervals not supported"));
                     }
                 }
             }
-            ArrowType::Float16 => Arc::new(
-                binary
-                    .iter()
-                    .map(|o| o.map(|b| f16::from_le_bytes(b[..2].try_into().unwrap())))
-                    .collect::<Float16Array>(),
-            ) as ArrayRef,
+            ArrowType::Float16 => {
+                let nulls = binary.nulls().cloned();
+                let f16s = binary.iter().map(|o| match o {
+                    Some(b) => f16::from_le_bytes(b[..2].try_into().unwrap()),
+                    None => f16::default(),
+                });
+                let f16s = Float16Array::from_iter_values_with_nulls(f16s, nulls);
+                Arc::new(f16s) as ArrayRef
+            }
             _ => Arc::new(binary) as ArrayRef,
         };
 
@@ -410,7 +421,6 @@ impl ColumnValueDecoder for ValueDecoder {
                 // so `offset` should be the value offset, not the byte offset
                 let total_values = buf.len() / self.byte_length;
                 let to_read = num_values.min(total_values - *offset);
-                out.buffer.reserve(to_read * self.byte_length);
 
                 // now read the n streams and reassemble values into the output buffer
                 read_byte_stream_split(&mut out.buffer, buf, *offset, to_read, self.byte_length);
@@ -453,9 +463,13 @@ fn read_byte_stream_split(
     data_width: usize,
 ) {
     let stride = src.len() / data_width;
-    for i in 0..num_values {
-        for j in 0..data_width {
-            dst.push(src[offset + j * stride + i]);
+    let idx = dst.len();
+    dst.resize(idx + num_values * data_width, 0u8);
+    let dst_slc = &mut dst[idx..idx + num_values * data_width];
+    for j in 0..data_width {
+        let src_slc = &src[offset + j * stride..offset + j * stride + num_values];
+        for i in 0..num_values {
+            dst_slc[i * data_width + j] = src_slc[i];
         }
     }
 }
diff --git a/parquet/src/arrow/array_reader/primitive_array.rs b/parquet/src/arrow/array_reader/primitive_array.rs
index 07ecc27d9..5e0e09212 100644
--- a/parquet/src/arrow/array_reader/primitive_array.rs
+++ b/parquet/src/arrow/array_reader/primitive_array.rs
@@ -217,22 +217,35 @@ where
                 arrow_cast::cast(&a, target_type)?
             }
             ArrowType::Decimal128(p, s) => {
+                // We can simply reuse the null buffer from `array` rather than recomputing it
+                // (as was the case when we simply used `collect` to produce the new array).
+                let nulls = array.nulls().cloned();
                 let array = match array.data_type() {
-                    ArrowType::Int32 => array
-                        .as_any()
-                        .downcast_ref::<Int32Array>()
-                        .unwrap()
-                        .iter()
-                        .map(|v| v.map(|v| v as i128))
-                        .collect::<Decimal128Array>(),
+                    ArrowType::Int32 => {
+                        let decimal = array
+                            .as_any()
+                            .downcast_ref::<Int32Array>()
+                            .unwrap()
+                            .iter()
+                            .map(|v| match v {
+                                Some(i) => i as i128,
+                                None => i128::default(),
+                            });
+                        Decimal128Array::from_iter_values_with_nulls(decimal, nulls)
+                    }
 
-                    ArrowType::Int64 => array
-                        .as_any()
-                        .downcast_ref::<Int64Array>()
-                        .unwrap()
-                        .iter()
-                        .map(|v| v.map(|v| v as i128))
-                        .collect::<Decimal128Array>(),
+                    ArrowType::Int64 => {
+                        let decimal = array
+                            .as_any()
+                            .downcast_ref::<Int64Array>()
+                            .unwrap()
+                            .iter()
+                            .map(|v| match v {
+                                Some(i) => i as i128,
+                                None => i128::default(),
+                            });
+                        Decimal128Array::from_iter_values_with_nulls(decimal, nulls)
+                    }
                     _ => {
                         return Err(arrow_err!(
                             "Cannot convert {:?} to decimal",
@@ -245,22 +258,35 @@ where
                 Arc::new(array) as ArrayRef
             }
             ArrowType::Decimal256(p, s) => {
+                // We can simply reuse the null buffer from `array` rather than recomputing it
+                // (as was the case when we simply used `collect` to produce the new array).
+                let nulls = array.nulls().cloned();
                 let array = match array.data_type() {
-                    ArrowType::Int32 => array
-                        .as_any()
-                        .downcast_ref::<Int32Array>()
-                        .unwrap()
-                        .iter()
-                        .map(|v| v.map(|v| i256::from_i128(v as i128)))
-                        .collect::<Decimal256Array>(),
+                    ArrowType::Int32 => {
+                        let decimal = array
+                            .as_any()
+                            .downcast_ref::<Int32Array>()
+                            .unwrap()
+                            .iter()
+                            .map(|v| match v {
+                                Some(i) => i256::from_i128(i as i128),
+                                None => i256::default(),
+                            });
+                        Decimal256Array::from_iter_values_with_nulls(decimal, nulls)
+                    }
 
-                    ArrowType::Int64 => array
-                        .as_any()
-                        .downcast_ref::<Int64Array>()
-                        .unwrap()
-                        .iter()
-                        .map(|v| v.map(|v| i256::from_i128(v as i128)))
-                        .collect::<Decimal256Array>(),
+                    ArrowType::Int64 => {
+                        let decimal = array
+                            .as_any()
+                            .downcast_ref::<Int64Array>()
+                            .unwrap()
+                            .iter()
+                            .map(|v| match v {
+                                Some(i) => i256::from_i128(i as i128),
+                                None => i256::default(),
+                            });
+                        Decimal256Array::from_iter_values_with_nulls(decimal, nulls)
+                    }
                     _ => {
                         return Err(arrow_err!(
                             "Cannot convert {:?} to decimal",
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs b/parquet/src/arrow/arrow_reader/statistics.rs
index d487967c2..602a9ad5e 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -116,7 +116,7 @@ macro_rules! make_stats_iterator {
                 let next = self.iter.next();
                 next.map(|x| {
                     x.and_then(|stats| match stats {
-                        $parquet_statistics_type(s) if stats.has_min_max_set() => Some(s.$func()),
+                        $parquet_statistics_type(s) => s.$func(),
                         _ => None,
                     })
                 })
@@ -131,45 +131,85 @@ macro_rules! make_stats_iterator {
 
 make_stats_iterator!(
     MinBooleanStatsIterator,
-    min,
+    min_opt,
     ParquetStatistics::Boolean,
     bool
 );
 make_stats_iterator!(
     MaxBooleanStatsIterator,
-    max,
+    max_opt,
     ParquetStatistics::Boolean,
     bool
 );
-make_stats_iterator!(MinInt32StatsIterator, min, ParquetStatistics::Int32, i32);
-make_stats_iterator!(MaxInt32StatsIterator, max, ParquetStatistics::Int32, i32);
-make_stats_iterator!(MinInt64StatsIterator, min, ParquetStatistics::Int64, i64);
-make_stats_iterator!(MaxInt64StatsIterator, max, ParquetStatistics::Int64, i64);
-make_stats_iterator!(MinFloatStatsIterator, min, ParquetStatistics::Float, f32);
-make_stats_iterator!(MaxFloatStatsIterator, max, ParquetStatistics::Float, f32);
-make_stats_iterator!(MinDoubleStatsIterator, min, ParquetStatistics::Double, f64);
-make_stats_iterator!(MaxDoubleStatsIterator, max, ParquetStatistics::Double, f64);
+make_stats_iterator!(
+    MinInt32StatsIterator,
+    min_opt,
+    ParquetStatistics::Int32,
+    i32
+);
+make_stats_iterator!(
+    MaxInt32StatsIterator,
+    max_opt,
+    ParquetStatistics::Int32,
+    i32
+);
+make_stats_iterator!(
+    MinInt64StatsIterator,
+    min_opt,
+    ParquetStatistics::Int64,
+    i64
+);
+make_stats_iterator!(
+    MaxInt64StatsIterator,
+    max_opt,
+    ParquetStatistics::Int64,
+    i64
+);
+make_stats_iterator!(
+    MinFloatStatsIterator,
+    min_opt,
+    ParquetStatistics::Float,
+    f32
+);
+make_stats_iterator!(
+    MaxFloatStatsIterator,
+    max_opt,
+    ParquetStatistics::Float,
+    f32
+);
+make_stats_iterator!(
+    MinDoubleStatsIterator,
+    min_opt,
+    ParquetStatistics::Double,
+    f64
+);
+make_stats_iterator!(
+    MaxDoubleStatsIterator,
+    max_opt,
+    ParquetStatistics::Double,
+    f64
+);
 make_stats_iterator!(
     MinByteArrayStatsIterator,
-    min_bytes,
+    min_bytes_opt,
     ParquetStatistics::ByteArray,
     [u8]
 );
 make_stats_iterator!(
     MaxByteArrayStatsIterator,
-    max_bytes,
+    max_bytes_opt,
     ParquetStatistics::ByteArray,
     [u8]
 );
 make_stats_iterator!(
     MinFixedLenByteArrayStatsIterator,
-    min_bytes,
+    min_bytes_opt,
     ParquetStatistics::FixedLenByteArray,
     [u8]
 );
 make_stats_iterator!(
     MaxFixedLenByteArrayStatsIterator,
-    max_bytes,
+    max_bytes_opt,
     ParquetStatistics::FixedLenByteArray,
     [u8]
 );
@@ -218,19 +258,18 @@ macro_rules! make_decimal_stats_iterator {
             fn next(&mut self) -> Option<Self::Item> {
                 let next = self.iter.next();
                 next.map(|x| {
-                    x.and_then(|stats| {
-                        if !stats.has_min_max_set() {
-                            return None;
+                    x.and_then(|stats| match stats {
+                        ParquetStatistics::Int32(s) => {
+                            s.$func().map(|x| $stat_value_type::from(*x))
                         }
-                        match stats {
-                            ParquetStatistics::Int32(s) => Some($stat_value_type::from(*s.$func())),
-                            ParquetStatistics::Int64(s) => Some($stat_value_type::from(*s.$func())),
-                            ParquetStatistics::ByteArray(s) => Some($convert_func(s.$bytes_func())),
-                            ParquetStatistics::FixedLenByteArray(s) => {
-                                Some($convert_func(s.$bytes_func()))
-                            }
-                            _ => None,
+                        ParquetStatistics::Int64(s) => {
+                            s.$func().map(|x| $stat_value_type::from(*x))
                         }
+                        ParquetStatistics::ByteArray(s) => s.$bytes_func().map($convert_func),
+                        ParquetStatistics::FixedLenByteArray(s) => {
+                            s.$bytes_func().map($convert_func)
+                        }
+                        _ => None,
                     })
                 })
             }
@@ -244,29 +283,29 @@ macro_rules! make_decimal_stats_iterator {
 
 make_decimal_stats_iterator!(
     MinDecimal128StatsIterator,
-    min,
-    min_bytes,
+    min_opt,
+    min_bytes_opt,
     i128,
     from_bytes_to_i128
 );
 make_decimal_stats_iterator!(
     MaxDecimal128StatsIterator,
-    max,
-    max_bytes,
+    max_opt,
+    max_bytes_opt,
     i128,
     from_bytes_to_i128
 );
 make_decimal_stats_iterator!(
     MinDecimal256StatsIterator,
-    min,
-    min_bytes,
+    min_opt,
+    min_bytes_opt,
     i256,
     from_bytes_to_i256
 );
 make_decimal_stats_iterator!(
     MaxDecimal256StatsIterator,
-    max,
-    max_bytes,
+    max_opt,
+    max_bytes_opt,
     i256,
     from_bytes_to_i256
 );
@@ -1347,7 +1386,7 @@ impl<'a> StatisticsConverter<'a> {
         let null_counts = metadatas
             .into_iter()
             .map(|x| x.column(parquet_index).statistics())
-            .map(|s| s.map(|s| s.null_count()));
+            .map(|s| s.and_then(|s| s.null_count_opt()));
         Ok(UInt64Array::from_iter(null_counts))
     }
 
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 0c07f541b..5c5f28ac0 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -2541,10 +2541,15 @@ mod tests {
                 row_offset += column.num_values() as usize;
 
                 let stats = column.statistics().unwrap();
-                assert!(stats.has_min_max_set());
                 if let Statistics::Int32(stats) = stats {
-                    assert_eq!(*stats.min() as u32, *src_slice.iter().min().unwrap());
-                    assert_eq!(*stats.max() as u32, *src_slice.iter().max().unwrap());
+                    assert_eq!(
+                        *stats.min_opt().unwrap() as u32,
+                        *src_slice.iter().min().unwrap()
+                    );
+                    assert_eq!(
+                        *stats.max_opt().unwrap() as u32,
+                        *src_slice.iter().max().unwrap()
+                    );
                 } else {
                     panic!("Statistics::Int32 missing")
                 }
@@ -2582,10 +2587,15 @@ mod tests {
                 row_offset += column.num_values() as usize;
 
                 let stats = column.statistics().unwrap();
-                assert!(stats.has_min_max_set());
                 if let Statistics::Int64(stats) = stats {
-                    assert_eq!(*stats.min() as u64, *src_slice.iter().min().unwrap());
-                    assert_eq!(*stats.max() as u64, *src_slice.iter().max().unwrap());
+                    assert_eq!(
+                        *stats.min_opt().unwrap() as u64,
+                        *src_slice.iter().min().unwrap()
+                    );
+                    assert_eq!(
+                        *stats.max_opt().unwrap() as u64,
+                        *src_slice.iter().max().unwrap()
+                    );
                 } else {
                     panic!("Statistics::Int64 missing")
                 }
@@ -2608,7 +2618,7 @@ mod tests {
             assert_eq!(row_group.num_columns(), 1);
             let column = row_group.column(0);
             let stats = column.statistics().unwrap();
-            assert_eq!(stats.null_count(), 2);
+            assert_eq!(stats.null_count_opt(), Some(2));
         }
     }
 
@@ -3069,8 +3079,8 @@ mod tests {
 
         // Column chunk of column "a" should have chunk level statistics
         if let Statistics::ByteArray(byte_array_stats) = a_col.statistics().unwrap() {
-            let min = byte_array_stats.min();
-            let max = byte_array_stats.max();
+            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']);
@@ -3141,8 +3151,8 @@ mod tests {
 
         // Column chunk of column "a" should have chunk level statistics
         if let Statistics::ByteArray(byte_array_stats) = a_col.statistics().unwrap() {
-            let min = byte_array_stats.min();
-            let max = byte_array_stats.max();
+            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']);
diff --git a/parquet/src/arrow/schema/complex.rs b/parquet/src/arrow/schema/complex.rs
index 3107a282e..a5f6d242e 100644
--- a/parquet/src/arrow/schema/complex.rs
+++ b/parquet/src/arrow/schema/complex.rs
@@ -38,10 +38,11 @@ fn get_repetition(t: &Type) -> Repetition {
 #[derive(Debug, Clone)]
 pub struct ParquetField {
     /// The level which represents an insertion into the current list
-    /// i.e. guaranteed to be > 0 for a list type
+    /// i.e. guaranteed to be > 0 for an element of list type
     pub rep_level: i16,
     /// The level at which this field is fully defined,
-    /// i.e. guaranteed to be > 0 for a nullable type
+    /// i.e. guaranteed to be > 0 for a nullable type or child of a 
+    /// nullable type
     pub def_level: i16,
     /// Whether this field is nullable
     pub nullable: bool,
diff --git a/parquet/src/column/page.rs b/parquet/src/column/page.rs
index 585d1951c..e3931dfe9 100644
--- a/parquet/src/column/page.rs
+++ b/parquet/src/column/page.rs
@@ -370,7 +370,7 @@ mod tests {
             encoding: Encoding::PLAIN,
             def_level_encoding: Encoding::RLE,
             rep_level_encoding: Encoding::RLE,
-            statistics: Some(Statistics::int32(Some(1), Some(2), None, 1, true)),
+            statistics: Some(Statistics::int32(Some(1), Some(2), None, Some(1), true)),
         };
         assert_eq!(data_page.page_type(), PageType::DATA_PAGE);
         assert_eq!(data_page.buffer(), vec![0, 1, 2].as_slice());
@@ -378,7 +378,7 @@ mod tests {
         assert_eq!(data_page.encoding(), Encoding::PLAIN);
         assert_eq!(
             data_page.statistics(),
-            Some(&Statistics::int32(Some(1), Some(2), None, 1, true))
+            Some(&Statistics::int32(Some(1), Some(2), None, Some(1), true))
         );
 
         let data_page_v2 = Page::DataPageV2 {
@@ -390,7 +390,7 @@ mod tests {
             def_levels_byte_len: 30,
             rep_levels_byte_len: 40,
             is_compressed: false,
-            statistics: Some(Statistics::int32(Some(1), Some(2), None, 1, true)),
+            statistics: Some(Statistics::int32(Some(1), Some(2), None, Some(1), true)),
         };
         assert_eq!(data_page_v2.page_type(), PageType::DATA_PAGE_V2);
         assert_eq!(data_page_v2.buffer(), vec![0, 1, 2].as_slice());
@@ -398,7 +398,7 @@ mod tests {
         assert_eq!(data_page_v2.encoding(), Encoding::PLAIN);
         assert_eq!(
             data_page_v2.statistics(),
-            Some(&Statistics::int32(Some(1), Some(2), None, 1, true))
+            Some(&Statistics::int32(Some(1), Some(2), None, Some(1), true))
         );
 
         let dict_page = Page::DictionaryPage {
@@ -422,7 +422,7 @@ mod tests {
             encoding: Encoding::PLAIN,
             def_level_encoding: Encoding::RLE,
             rep_level_encoding: Encoding::RLE,
-            statistics: Some(Statistics::int32(Some(1), Some(2), None, 1, true)),
+            statistics: Some(Statistics::int32(Some(1), Some(2), None, Some(1), true)),
         };
 
         let cpage = CompressedPage::new(data_page, 5);
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 519a219f9..8ea287831 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -769,8 +769,8 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
                 }
                 Some(stat) => {
                     // Check if min/max are still ascending/descending across pages
-                    let new_min = stat.min();
-                    let new_max = stat.max();
+                    let new_min = stat.min_opt().unwrap();
+                    let new_max = stat.max_opt().unwrap();
                     if let Some((last_min, last_max)) = &self.last_non_null_data_page_min_max {
                         if self.data_page_boundary_ascending {
                             // If last min/max are greater than new min/max then not ascending anymore
@@ -797,12 +797,12 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
                             null_page,
                             self.truncate_min_value(
                                 self.props.column_index_truncate_length(),
-                                stat.min_bytes(),
+                                stat.min_bytes_opt().unwrap(),
                             )
                             .0,
                             self.truncate_max_value(
                                 self.props.column_index_truncate_length(),
-                                stat.max_bytes(),
+                                stat.max_bytes_opt().unwrap(),
                             )
                             .0,
                             self.page_metrics.num_page_nulls as i64,
@@ -810,8 +810,8 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
                     } else {
                         self.column_index_builder.append(
                             null_page,
-                            stat.min_bytes().to_vec(),
-                            stat.max_bytes().to_vec(),
+                            stat.min_bytes_opt().unwrap().to_vec(),
+                            stat.max_bytes_opt().unwrap().to_vec(),
                             self.page_metrics.num_page_nulls as i64,
                         );
                     }
@@ -897,7 +897,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
                         Some(min),
                         Some(max),
                         None,
-                        self.page_metrics.num_page_nulls,
+                        Some(self.page_metrics.num_page_nulls),
                         false,
                     ),
                 )
@@ -1066,28 +1066,28 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
                 self.column_metrics.min_column_value.clone(),
                 self.column_metrics.max_column_value.clone(),
                 self.column_metrics.column_distinct_count,
-                self.column_metrics.num_column_nulls,
+                Some(self.column_metrics.num_column_nulls),
                 false,
             )
             .with_backwards_compatible_min_max(backwards_compatible_min_max)
             .into();
 
             let statistics = match statistics {
-                Statistics::ByteArray(stats) if stats.has_min_max_set() => {
+                Statistics::ByteArray(stats) if stats._internal_has_min_max_set() => {
                     let (min, did_truncate_min) = self.truncate_min_value(
                         self.props.statistics_truncate_length(),
-                        stats.min_bytes(),
+                        stats.min_bytes_opt().unwrap(),
                     );
                     let (max, did_truncate_max) = self.truncate_max_value(
                         self.props.statistics_truncate_length(),
-                        stats.max_bytes(),
+                        stats.max_bytes_opt().unwrap(),
                     );
                     Statistics::ByteArray(
                         ValueStatistics::new(
                             Some(min.into()),
                             Some(max.into()),
                             stats.distinct_count(),
-                            stats.null_count(),
+                            stats.null_count_opt(),
                             backwards_compatible_min_max,
                         )
                         .with_max_is_exact(!did_truncate_max)
@@ -1095,22 +1095,22 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
                     )
                 }
                 Statistics::FixedLenByteArray(stats)
-                    if (stats.has_min_max_set() && self.can_truncate_value()) =>
+                    if (stats._internal_has_min_max_set() && self.can_truncate_value()) =>
                 {
                     let (min, did_truncate_min) = self.truncate_min_value(
                         self.props.statistics_truncate_length(),
-                        stats.min_bytes(),
+                        stats.min_bytes_opt().unwrap(),
                     );
                     let (max, did_truncate_max) = self.truncate_max_value(
                         self.props.statistics_truncate_length(),
-                        stats.max_bytes(),
+                        stats.max_bytes_opt().unwrap(),
                     );
                     Statistics::FixedLenByteArray(
                         ValueStatistics::new(
                             Some(min.into()),
                             Some(max.into()),
                             stats.distinct_count(),
-                            stats.null_count(),
+                            stats.null_count_opt(),
                             backwards_compatible_min_max,
                         )
                         .with_max_is_exact(!did_truncate_max)
@@ -1841,12 +1841,11 @@ mod tests {
         assert_eq!(metadata.data_page_offset(), 0);
         assert_eq!(metadata.dictionary_page_offset(), Some(0));
         if let Some(stats) = metadata.statistics() {
-            assert!(stats.has_min_max_set());
-            assert_eq!(stats.null_count(), 0);
+            assert_eq!(stats.null_count_opt(), Some(0));
             assert_eq!(stats.distinct_count(), None);
             if let Statistics::Int32(stats) = stats {
-                assert_eq!(stats.min(), &1);
-                assert_eq!(stats.max(), &4);
+                assert_eq!(stats.min_opt().unwrap(), &1);
+                assert_eq!(stats.max_opt().unwrap(), &4);
             } else {
                 panic!("expecting Statistics::Int32");
             }
@@ -1886,17 +1885,16 @@ mod tests {
             .unwrap();
         let metadata = writer.close().unwrap().metadata;
         if let Some(stats) = metadata.statistics() {
-            assert!(stats.has_min_max_set());
             if let Statistics::ByteArray(stats) = stats {
                 assert_eq!(
-                    stats.min(),
+                    stats.min_opt().unwrap(),
                     &ByteArray::from(vec![
                         255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 179u8, 172u8, 19u8,
                         35u8, 231u8, 90u8, 0u8, 0u8,
                     ])
                 );
                 assert_eq!(
-                    stats.max(),
+                    stats.max_opt().unwrap(),
                     &ByteArray::from(vec![
                         0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 41u8, 162u8, 36u8, 26u8, 246u8,
                         44u8, 0u8, 0u8,
@@ -1923,10 +1921,9 @@ mod tests {
         writer.write_batch(&[0, 1, 2, 3, 4, 5], None, None).unwrap();
         let metadata = writer.close().unwrap().metadata;
         if let Some(stats) = metadata.statistics() {
-            assert!(stats.has_min_max_set());
             if let Statistics::Int32(stats) = stats {
-                assert_eq!(stats.min(), &0,);
-                assert_eq!(stats.max(), &5,);
+                assert_eq!(stats.min_opt().unwrap(), &0,);
+                assert_eq!(stats.max_opt().unwrap(), &5,);
             } else {
                 panic!("expecting Statistics::Int32");
             }
@@ -1970,12 +1967,11 @@ mod tests {
         assert_eq!(metadata.data_page_offset(), 0);
         assert_eq!(metadata.dictionary_page_offset(), Some(0));
         if let Some(stats) = metadata.statistics() {
-            assert!(stats.has_min_max_set());
-            assert_eq!(stats.null_count(), 0);
+            assert_eq!(stats.null_count_opt(), Some(0));
             assert_eq!(stats.distinct_count().unwrap_or(0), 55);
             if let Statistics::Int32(stats) = stats {
-                assert_eq!(stats.min(), &-17);
-                assert_eq!(stats.max(), &9000);
+                assert_eq!(stats.min_opt().unwrap(), &-17);
+                assert_eq!(stats.max_opt().unwrap(), &9000);
             } else {
                 panic!("expecting Statistics::Int32");
             }
@@ -2000,9 +1996,9 @@ mod tests {
         let r = writer.close().unwrap();
 
         let stats = r.metadata.statistics().unwrap();
-        assert_eq!(stats.min_bytes(), 1_i32.to_le_bytes());
-        assert_eq!(stats.max_bytes(), 7_i32.to_le_bytes());
-        assert_eq!(stats.null_count(), 0);
+        assert_eq!(stats.min_bytes_opt().unwrap(), 1_i32.to_le_bytes());
+        assert_eq!(stats.max_bytes_opt().unwrap(), 7_i32.to_le_bytes());
+        assert_eq!(stats.null_count_opt(), Some(0));
         assert!(stats.distinct_count().is_none());
 
         drop(write);
@@ -2026,9 +2022,15 @@ mod tests {
         assert_eq!(pages[1].page_type(), PageType::DATA_PAGE);
 
         let page_statistics = pages[1].statistics().unwrap();
-        assert_eq!(page_statistics.min_bytes(), 1_i32.to_le_bytes());
-        assert_eq!(page_statistics.max_bytes(), 7_i32.to_le_bytes());
-        assert_eq!(page_statistics.null_count(), 0);
+        assert_eq!(
+            page_statistics.min_bytes_opt().unwrap(),
+            1_i32.to_le_bytes()
+        );
+        assert_eq!(
+            page_statistics.max_bytes_opt().unwrap(),
+            7_i32.to_le_bytes()
+        );
+        assert_eq!(page_statistics.null_count_opt(), Some(0));
         assert!(page_statistics.distinct_count().is_none());
     }
 
@@ -2217,13 +2219,12 @@ mod tests {
     #[test]
     fn test_bool_statistics() {
         let stats = statistics_roundtrip::<BoolType>(&[true, false, false, true]);
-        assert!(stats.has_min_max_set());
         // Booleans have an unsigned sort order and so are not compatible
         // with the deprecated `min` and `max` statistics
         assert!(!stats.is_min_max_backwards_compatible());
         if let Statistics::Boolean(stats) = stats {
-            assert_eq!(stats.min(), &false);
-            assert_eq!(stats.max(), &true);
+            assert_eq!(stats.min_opt().unwrap(), &false);
+            assert_eq!(stats.max_opt().unwrap(), &true);
         } else {
             panic!("expecting Statistics::Boolean, got {stats:?}");
         }
@@ -2232,11 +2233,10 @@ mod tests {
     #[test]
     fn test_int32_statistics() {
         let stats = statistics_roundtrip::<Int32Type>(&[-1, 3, -2, 2]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Int32(stats) = stats {
-            assert_eq!(stats.min(), &-2);
-            assert_eq!(stats.max(), &3);
+            assert_eq!(stats.min_opt().unwrap(), &-2);
+            assert_eq!(stats.max_opt().unwrap(), &3);
         } else {
             panic!("expecting Statistics::Int32, got {stats:?}");
         }
@@ -2245,11 +2245,10 @@ mod tests {
     #[test]
     fn test_int64_statistics() {
         let stats = statistics_roundtrip::<Int64Type>(&[-1, 3, -2, 2]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Int64(stats) = stats {
-            assert_eq!(stats.min(), &-2);
-            assert_eq!(stats.max(), &3);
+            assert_eq!(stats.min_opt().unwrap(), &-2);
+            assert_eq!(stats.max_opt().unwrap(), &3);
         } else {
             panic!("expecting Statistics::Int64, got {stats:?}");
         }
@@ -2267,11 +2266,10 @@ mod tests {
         .collect::<Vec<Int96>>();
 
         let stats = statistics_roundtrip::<Int96Type>(&input);
-        assert!(stats.has_min_max_set());
         assert!(!stats.is_min_max_backwards_compatible());
         if let Statistics::Int96(stats) = stats {
-            assert_eq!(stats.min(), &Int96::from(vec![0, 20, 30]));
-            assert_eq!(stats.max(), &Int96::from(vec![3, 20, 10]));
+            assert_eq!(stats.min_opt().unwrap(), &Int96::from(vec![0, 20, 30]));
+            assert_eq!(stats.max_opt().unwrap(), &Int96::from(vec![3, 20, 10]));
         } else {
             panic!("expecting Statistics::Int96, got {stats:?}");
         }
@@ -2280,11 +2278,10 @@ mod tests {
     #[test]
     fn test_float_statistics() {
         let stats = statistics_roundtrip::<FloatType>(&[-1.0, 3.0, -2.0, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &-2.0);
-            assert_eq!(stats.max(), &3.0);
+            assert_eq!(stats.min_opt().unwrap(), &-2.0);
+            assert_eq!(stats.max_opt().unwrap(), &3.0);
         } else {
             panic!("expecting Statistics::Float, got {stats:?}");
         }
@@ -2293,11 +2290,10 @@ mod tests {
     #[test]
     fn test_double_statistics() {
         let stats = statistics_roundtrip::<DoubleType>(&[-1.0, 3.0, -2.0, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &-2.0);
-            assert_eq!(stats.max(), &3.0);
+            assert_eq!(stats.min_opt().unwrap(), &-2.0);
+            assert_eq!(stats.max_opt().unwrap(), &3.0);
         } else {
             panic!("expecting Statistics::Double, got {stats:?}");
         }
@@ -2312,10 +2308,9 @@ mod tests {
 
         let stats = statistics_roundtrip::<ByteArrayType>(&input);
         assert!(!stats.is_min_max_backwards_compatible());
-        assert!(stats.has_min_max_set());
         if let Statistics::ByteArray(stats) = stats {
-            assert_eq!(stats.min(), &ByteArray::from("aaw"));
-            assert_eq!(stats.max(), &ByteArray::from("zz"));
+            assert_eq!(stats.min_opt().unwrap(), &ByteArray::from("aaw"));
+            assert_eq!(stats.max_opt().unwrap(), &ByteArray::from("zz"));
         } else {
             panic!("expecting Statistics::ByteArray, got {stats:?}");
         }
@@ -2329,13 +2324,12 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = statistics_roundtrip::<FixedLenByteArrayType>(&input);
-        assert!(stats.has_min_max_set());
         assert!(!stats.is_min_max_backwards_compatible());
         if let Statistics::FixedLenByteArray(stats) = stats {
             let expected_min: FixedLenByteArray = ByteArray::from("aaw  ").into();
-            assert_eq!(stats.min(), &expected_min);
+            assert_eq!(stats.min_opt().unwrap(), &expected_min);
             let expected_max: FixedLenByteArray = ByteArray::from("zz   ").into();
-            assert_eq!(stats.max(), &expected_max);
+            assert_eq!(stats.max_opt().unwrap(), &expected_max);
         } else {
             panic!("expecting Statistics::FixedLenByteArray, got {stats:?}");
         }
@@ -2354,10 +2348,15 @@ mod tests {
         .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(-f16::from_f32(2.0)));
-        assert_eq!(stats.max(), &ByteArray::from(f16::from_f32(3.0)));
+        assert_eq!(
+            stats.min_opt().unwrap(),
+            &ByteArray::from(-f16::from_f32(2.0))
+        );
+        assert_eq!(
+            stats.max_opt().unwrap(),
+            &ByteArray::from(f16::from_f32(3.0))
+        );
     }
 
     #[test]
@@ -2368,10 +2367,12 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(f16::ONE));
-        assert_eq!(stats.max(), &ByteArray::from(f16::ONE + f16::ONE));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
+        assert_eq!(
+            stats.max_opt().unwrap(),
+            &ByteArray::from(f16::ONE + f16::ONE)
+        );
     }
 
     #[test]
@@ -2382,10 +2383,12 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(f16::ONE));
-        assert_eq!(stats.max(), &ByteArray::from(f16::ONE + f16::ONE));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
+        assert_eq!(
+            stats.max_opt().unwrap(),
+            &ByteArray::from(f16::ONE + f16::ONE)
+        );
     }
 
     #[test]
@@ -2396,10 +2399,12 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(f16::ONE));
-        assert_eq!(stats.max(), &ByteArray::from(f16::ONE + f16::ONE));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
+        assert_eq!(
+            stats.max_opt().unwrap(),
+            &ByteArray::from(f16::ONE + f16::ONE)
+        );
     }
 
     #[test]
@@ -2410,7 +2415,8 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(!stats.has_min_max_set());
+        assert!(stats.min_bytes_opt().is_none());
+        assert!(stats.max_bytes_opt().is_none());
         assert!(stats.is_min_max_backwards_compatible());
     }
 
@@ -2422,10 +2428,9 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(f16::NEG_ZERO));
-        assert_eq!(stats.max(), &ByteArray::from(f16::ZERO));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
+        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
     }
 
     #[test]
@@ -2436,10 +2441,9 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(f16::NEG_ZERO));
-        assert_eq!(stats.max(), &ByteArray::from(f16::ZERO));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
+        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
     }
 
     #[test]
@@ -2450,10 +2454,9 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(f16::NEG_ZERO));
-        assert_eq!(stats.max(), &ByteArray::from(f16::PI));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
+        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::PI));
     }
 
     #[test]
@@ -2464,20 +2467,18 @@ mod tests {
             .collect::<Vec<_>>();
 
         let stats = float16_statistics_roundtrip(&input);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
-        assert_eq!(stats.min(), &ByteArray::from(-f16::PI));
-        assert_eq!(stats.max(), &ByteArray::from(f16::ZERO));
+        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(-f16::PI));
+        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
     }
 
     #[test]
     fn test_float_statistics_nan_middle() {
         let stats = statistics_roundtrip::<FloatType>(&[1.0, f32::NAN, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &1.0);
-            assert_eq!(stats.max(), &2.0);
+            assert_eq!(stats.min_opt().unwrap(), &1.0);
+            assert_eq!(stats.max_opt().unwrap(), &2.0);
         } else {
             panic!("expecting Statistics::Float");
         }
@@ -2486,11 +2487,10 @@ mod tests {
     #[test]
     fn test_float_statistics_nan_start() {
         let stats = statistics_roundtrip::<FloatType>(&[f32::NAN, 1.0, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &1.0);
-            assert_eq!(stats.max(), &2.0);
+            assert_eq!(stats.min_opt().unwrap(), &1.0);
+            assert_eq!(stats.max_opt().unwrap(), &2.0);
         } else {
             panic!("expecting Statistics::Float");
         }
@@ -2499,7 +2499,8 @@ mod tests {
     #[test]
     fn test_float_statistics_nan_only() {
         let stats = statistics_roundtrip::<FloatType>(&[f32::NAN, f32::NAN]);
-        assert!(!stats.has_min_max_set());
+        assert!(stats.min_bytes_opt().is_none());
+        assert!(stats.max_bytes_opt().is_none());
         assert!(stats.is_min_max_backwards_compatible());
         assert!(matches!(stats, Statistics::Float(_)));
     }
@@ -2507,13 +2508,12 @@ mod tests {
     #[test]
     fn test_float_statistics_zero_only() {
         let stats = statistics_roundtrip::<FloatType>(&[0.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &-0.0);
-            assert!(stats.min().is_sign_negative());
-            assert_eq!(stats.max(), &0.0);
-            assert!(stats.max().is_sign_positive());
+            assert_eq!(stats.min_opt().unwrap(), &-0.0);
+            assert!(stats.min_opt().unwrap().is_sign_negative());
+            assert_eq!(stats.max_opt().unwrap(), &0.0);
+            assert!(stats.max_opt().unwrap().is_sign_positive());
         } else {
             panic!("expecting Statistics::Float");
         }
@@ -2522,13 +2522,12 @@ mod tests {
     #[test]
     fn test_float_statistics_neg_zero_only() {
         let stats = statistics_roundtrip::<FloatType>(&[-0.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &-0.0);
-            assert!(stats.min().is_sign_negative());
-            assert_eq!(stats.max(), &0.0);
-            assert!(stats.max().is_sign_positive());
+            assert_eq!(stats.min_opt().unwrap(), &-0.0);
+            assert!(stats.min_opt().unwrap().is_sign_negative());
+            assert_eq!(stats.max_opt().unwrap(), &0.0);
+            assert!(stats.max_opt().unwrap().is_sign_positive());
         } else {
             panic!("expecting Statistics::Float");
         }
@@ -2537,12 +2536,11 @@ mod tests {
     #[test]
     fn test_float_statistics_zero_min() {
         let stats = statistics_roundtrip::<FloatType>(&[0.0, 1.0, f32::NAN, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &-0.0);
-            assert!(stats.min().is_sign_negative());
-            assert_eq!(stats.max(), &2.0);
+            assert_eq!(stats.min_opt().unwrap(), &-0.0);
+            assert!(stats.min_opt().unwrap().is_sign_negative());
+            assert_eq!(stats.max_opt().unwrap(), &2.0);
         } else {
             panic!("expecting Statistics::Float");
         }
@@ -2551,12 +2549,11 @@ mod tests {
     #[test]
     fn test_float_statistics_neg_zero_max() {
         let stats = statistics_roundtrip::<FloatType>(&[-0.0, -1.0, f32::NAN, -2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Float(stats) = stats {
-            assert_eq!(stats.min(), &-2.0);
-            assert_eq!(stats.max(), &0.0);
-            assert!(stats.max().is_sign_positive());
+            assert_eq!(stats.min_opt().unwrap(), &-2.0);
+            assert_eq!(stats.max_opt().unwrap(), &0.0);
+            assert!(stats.max_opt().unwrap().is_sign_positive());
         } else {
             panic!("expecting Statistics::Float");
         }
@@ -2565,11 +2562,10 @@ mod tests {
     #[test]
     fn test_double_statistics_nan_middle() {
         let stats = statistics_roundtrip::<DoubleType>(&[1.0, f64::NAN, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &1.0);
-            assert_eq!(stats.max(), &2.0);
+            assert_eq!(stats.min_opt().unwrap(), &1.0);
+            assert_eq!(stats.max_opt().unwrap(), &2.0);
         } else {
             panic!("expecting Statistics::Double");
         }
@@ -2578,11 +2574,10 @@ mod tests {
     #[test]
     fn test_double_statistics_nan_start() {
         let stats = statistics_roundtrip::<DoubleType>(&[f64::NAN, 1.0, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &1.0);
-            assert_eq!(stats.max(), &2.0);
+            assert_eq!(stats.min_opt().unwrap(), &1.0);
+            assert_eq!(stats.max_opt().unwrap(), &2.0);
         } else {
             panic!("expecting Statistics::Double");
         }
@@ -2591,7 +2586,8 @@ mod tests {
     #[test]
     fn test_double_statistics_nan_only() {
         let stats = statistics_roundtrip::<DoubleType>(&[f64::NAN, f64::NAN]);
-        assert!(!stats.has_min_max_set());
+        assert!(stats.min_bytes_opt().is_none());
+        assert!(stats.max_bytes_opt().is_none());
         assert!(matches!(stats, Statistics::Double(_)));
         assert!(stats.is_min_max_backwards_compatible());
     }
@@ -2599,13 +2595,12 @@ mod tests {
     #[test]
     fn test_double_statistics_zero_only() {
         let stats = statistics_roundtrip::<DoubleType>(&[0.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &-0.0);
-            assert!(stats.min().is_sign_negative());
-            assert_eq!(stats.max(), &0.0);
-            assert!(stats.max().is_sign_positive());
+            assert_eq!(stats.min_opt().unwrap(), &-0.0);
+            assert!(stats.min_opt().unwrap().is_sign_negative());
+            assert_eq!(stats.max_opt().unwrap(), &0.0);
+            assert!(stats.max_opt().unwrap().is_sign_positive());
         } else {
             panic!("expecting Statistics::Double");
         }
@@ -2614,13 +2609,12 @@ mod tests {
     #[test]
     fn test_double_statistics_neg_zero_only() {
         let stats = statistics_roundtrip::<DoubleType>(&[-0.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &-0.0);
-            assert!(stats.min().is_sign_negative());
-            assert_eq!(stats.max(), &0.0);
-            assert!(stats.max().is_sign_positive());
+            assert_eq!(stats.min_opt().unwrap(), &-0.0);
+            assert!(stats.min_opt().unwrap().is_sign_negative());
+            assert_eq!(stats.max_opt().unwrap(), &0.0);
+            assert!(stats.max_opt().unwrap().is_sign_positive());
         } else {
             panic!("expecting Statistics::Double");
         }
@@ -2629,12 +2623,11 @@ mod tests {
     #[test]
     fn test_double_statistics_zero_min() {
         let stats = statistics_roundtrip::<DoubleType>(&[0.0, 1.0, f64::NAN, 2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &-0.0);
-            assert!(stats.min().is_sign_negative());
-            assert_eq!(stats.max(), &2.0);
+            assert_eq!(stats.min_opt().unwrap(), &-0.0);
+            assert!(stats.min_opt().unwrap().is_sign_negative());
+            assert_eq!(stats.max_opt().unwrap(), &2.0);
         } else {
             panic!("expecting Statistics::Double");
         }
@@ -2643,12 +2636,11 @@ mod tests {
     #[test]
     fn test_double_statistics_neg_zero_max() {
         let stats = statistics_roundtrip::<DoubleType>(&[-0.0, -1.0, f64::NAN, -2.0]);
-        assert!(stats.has_min_max_set());
         assert!(stats.is_min_max_backwards_compatible());
         if let Statistics::Double(stats) = stats {
-            assert_eq!(stats.min(), &-2.0);
-            assert_eq!(stats.max(), &0.0);
-            assert!(stats.max().is_sign_positive());
+            assert_eq!(stats.min_opt().unwrap(), &-2.0);
+            assert_eq!(stats.max_opt().unwrap(), &0.0);
+            assert!(stats.max_opt().unwrap().is_sign_positive());
         } else {
             panic!("expecting Statistics::Double");
         }
@@ -2705,15 +2697,20 @@ mod tests {
         }
 
         if let Some(stats) = r.metadata.statistics() {
-            assert!(stats.has_min_max_set());
-            assert_eq!(stats.null_count(), 0);
+            assert_eq!(stats.null_count_opt(), Some(0));
             assert_eq!(stats.distinct_count(), None);
             if let Statistics::Int32(stats) = stats {
                 // first page is [1,2,3,4]
                 // second page is [-5,2,4,8]
                 // note that we don't increment here, as this is a non BinaryArray type.
-                assert_eq!(stats.min_bytes(), column_index.min_values[1].as_slice());
-                assert_eq!(stats.max_bytes(), column_index.max_values.get(1).unwrap());
+                assert_eq!(
+                    stats.min_bytes_opt(),
+                    Some(column_index.min_values[1].as_slice())
+                );
+                assert_eq!(
+                    stats.max_bytes_opt(),
+                    column_index.max_values.get(1).map(Vec::as_slice)
+                );
             } else {
                 panic!("expecting Statistics::Int32");
             }
@@ -2760,16 +2757,21 @@ mod tests {
         assert_eq!(0, column_index.null_counts.as_ref().unwrap()[0]);
 
         if let Some(stats) = r.metadata.statistics() {
-            assert!(stats.has_min_max_set());
-            assert_eq!(stats.null_count(), 0);
+            assert_eq!(stats.null_count_opt(), Some(0));
             assert_eq!(stats.distinct_count(), None);
             if let Statistics::FixedLenByteArray(stats) = stats {
                 let column_index_min_value = &column_index.min_values[0];
                 let column_index_max_value = &column_index.max_values[0];
 
                 // Column index stats are truncated, while the column chunk's aren't.
-                assert_ne!(stats.min_bytes(), column_index_min_value.as_slice());
-                assert_ne!(stats.max_bytes(), column_index_max_value.as_slice());
+                assert_ne!(
+                    stats.min_bytes_opt(),
+                    Some(column_index_min_value.as_slice())
+                );
+                assert_ne!(
+                    stats.max_bytes_opt(),
+                    Some(column_index_max_value.as_slice())
+                );
 
                 assert_eq!(
                     column_index_min_value.len(),
@@ -2827,8 +2829,7 @@ mod tests {
         assert_eq!(0, column_index.null_counts.as_ref().unwrap()[0]);
 
         if let Some(stats) = r.metadata.statistics() {
-            assert!(stats.has_min_max_set());
-            assert_eq!(stats.null_count(), 0);
+            assert_eq!(stats.null_count_opt(), Some(0));
             assert_eq!(stats.distinct_count(), None);
             if let Statistics::FixedLenByteArray(_stats) = stats {
                 let column_index_min_value = &column_index.min_values[0];
@@ -2840,8 +2841,8 @@ mod tests {
                 assert_eq!("B".as_bytes(), column_index_min_value.as_slice());
                 assert_eq!("C".as_bytes(), column_index_max_value.as_slice());
 
-                assert_ne!(column_index_min_value, stats.min_bytes());
-                assert_ne!(column_index_max_value, stats.max_bytes());
+                assert_ne!(column_index_min_value, stats.min_bytes_opt().unwrap());
+                assert_ne!(column_index_max_value, stats.max_bytes_opt().unwrap());
             } else {
                 panic!("expecting Statistics::FixedLenByteArray");
             }
@@ -2875,10 +2876,9 @@ mod tests {
 
         // ensure bytes weren't truncated for statistics
         let stats = r.metadata.statistics().unwrap();
-        assert!(stats.has_min_max_set());
         if let Statistics::FixedLenByteArray(stats) = stats {
-            let stats_min_bytes = stats.min_bytes();
-            let stats_max_bytes = stats.max_bytes();
+            let stats_min_bytes = stats.min_bytes_opt().unwrap();
+            let stats_max_bytes = stats.max_bytes_opt().unwrap();
             assert_eq!(expected_value, stats_min_bytes);
             assert_eq!(expected_value, stats_max_bytes);
         } else {
@@ -2915,10 +2915,9 @@ mod tests {
 
         // ensure bytes weren't truncated for statistics
         let stats = r.metadata.statistics().unwrap();
-        assert!(stats.has_min_max_set());
         if let Statistics::FixedLenByteArray(stats) = stats {
-            let stats_min_bytes = stats.min_bytes();
-            let stats_max_bytes = stats.max_bytes();
+            let stats_min_bytes = stats.min_bytes_opt().unwrap();
+            let stats_max_bytes = stats.max_bytes_opt().unwrap();
             assert_eq!(expected_value, stats_min_bytes);
             assert_eq!(expected_value, stats_max_bytes);
         } else {
@@ -2951,12 +2950,11 @@ mod tests {
         assert_eq!(1, r.rows_written);
 
         let stats = r.metadata.statistics().expect("statistics");
-        assert!(stats.has_min_max_set());
-        assert_eq!(stats.null_count(), 0);
+        assert_eq!(stats.null_count_opt(), Some(0));
         assert_eq!(stats.distinct_count(), None);
         if let Statistics::ByteArray(_stats) = stats {
-            let min_value = _stats.min();
-            let max_value = _stats.max();
+            let min_value = _stats.min_opt().unwrap();
+            let max_value = _stats.max_opt().unwrap();
 
             assert!(!_stats.min_is_exact());
             assert!(!_stats.max_is_exact());
@@ -3004,12 +3002,11 @@ mod tests {
         assert_eq!(1, r.rows_written);
 
         let stats = r.metadata.statistics().expect("statistics");
-        assert!(stats.has_min_max_set());
-        assert_eq!(stats.null_count(), 0);
+        assert_eq!(stats.null_count_opt(), Some(0));
         assert_eq!(stats.distinct_count(), None);
         if let Statistics::FixedLenByteArray(_stats) = stats {
-            let min_value = _stats.min();
-            let max_value = _stats.max();
+            let min_value = _stats.min_opt().unwrap();
+            let max_value = _stats.max_opt().unwrap();
 
             assert!(!_stats.min_is_exact());
             assert!(!_stats.max_is_exact());
@@ -3326,7 +3323,8 @@ mod tests {
         } else {
             panic!("metadata missing statistics");
         };
-        assert!(!stats.has_min_max_set());
+        assert!(stats.min_bytes_opt().is_none());
+        assert!(stats.max_bytes_opt().is_none());
     }
 
     fn write_multiple_pages<T: DataType>(
diff --git a/parquet/src/data_type.rs b/parquet/src/data_type.rs
index 01e92115c..324e1c379 100644
--- a/parquet/src/data_type.rs
+++ b/parquet/src/data_type.rs
@@ -674,6 +674,13 @@ pub(crate) mod private {
 
         /// Return the value as an mutable Any to allow for downcasts without transmutation
         fn as_mut_any(&mut self) -> &mut dyn std::any::Any;
+
+        /// Sets the value of this object from the provided [`Bytes`]
+        ///
+        /// Only implemented for `ByteArray` and `FixedLenByteArray`. Will panic for other types.
+        fn set_from_bytes(&mut self, _data: Bytes) {
+            unimplemented!();
+        }
     }
 
     impl ParquetValueType for bool {
@@ -953,9 +960,7 @@ pub(crate) mod private {
                     return Err(eof_err!("Not enough bytes to decode"));
                 }
 
-                let val: &mut Self = val_array.as_mut_any().downcast_mut().unwrap();
-
-                val.set_data(data.slice(decoder.start..decoder.start + len));
+                val_array.set_data(data.slice(decoder.start..decoder.start + len));
                 decoder.start += len;
             }
             decoder.num_values -= num_values;
@@ -998,6 +1003,11 @@ pub(crate) mod private {
         fn as_mut_any(&mut self) -> &mut dyn std::any::Any {
             self
         }
+
+        #[inline]
+        fn set_from_bytes(&mut self, data: Bytes) {
+            self.set_data(data);
+        }
     }
 
     impl HeapSize for super::ByteArray {
@@ -1093,6 +1103,11 @@ pub(crate) mod private {
         fn as_mut_any(&mut self) -> &mut dyn std::any::Any {
             self
         }
+
+        #[inline]
+        fn set_from_bytes(&mut self, data: Bytes) {
+            self.set_data(data);
+        }
     }
 
     impl HeapSize for super::FixedLenByteArray {
diff --git a/parquet/src/encodings/decoding.rs b/parquet/src/encodings/decoding.rs
index 16467b32d..b5217d02f 100644
--- a/parquet/src/encodings/decoding.rs
+++ b/parquet/src/encodings/decoding.rs
@@ -901,11 +901,7 @@ impl<T: DataType> Decoder<T> for DeltaLengthByteArrayDecoder<T> {
 
                 for item in buffer.iter_mut().take(num_values) {
                     let len = self.lengths[self.current_idx] as usize;
-
-                    item.as_mut_any()
-                        .downcast_mut::<ByteArray>()
-                        .unwrap()
-                        .set_data(data.slice(self.offset..self.offset + len));
+                    item.set_from_bytes(data.slice(self.offset..self.offset + len));
 
                     self.offset += len;
                     self.current_idx += 1;
@@ -1029,7 +1025,7 @@ impl<T: DataType> Decoder<T> for DeltaByteArrayDecoder<T> {
 
     fn get(&mut self, buffer: &mut [T::T]) -> Result<usize> {
         match T::get_physical_type() {
-            ty @ Type::BYTE_ARRAY | ty @ Type::FIXED_LEN_BYTE_ARRAY => {
+            Type::BYTE_ARRAY | Type::FIXED_LEN_BYTE_ARRAY => {
                 let num_values = cmp::min(buffer.len(), self.num_values);
                 let mut v: [ByteArray; 1] = [ByteArray::new(); 1];
                 for item in buffer.iter_mut().take(num_values) {
@@ -1051,20 +1047,7 @@ impl<T: DataType> Decoder<T> for DeltaByteArrayDecoder<T> {
                     result.extend_from_slice(suffix);
 
                     let data = Bytes::from(result.clone());
-
-                    match ty {
-                        Type::BYTE_ARRAY => item
-                            .as_mut_any()
-                            .downcast_mut::<ByteArray>()
-                            .unwrap()
-                            .set_data(data),
-                        Type::FIXED_LEN_BYTE_ARRAY => item
-                            .as_mut_any()
-                            .downcast_mut::<FixedLenByteArray>()
-                            .unwrap()
-                            .set_data(data),
-                        _ => unreachable!(),
-                    };
+                    item.set_from_bytes(data);
 
                     self.previous_value = result;
                     self.current_idx += 1;
diff --git a/parquet/src/encodings/decoding/byte_stream_split_decoder.rs b/parquet/src/encodings/decoding/byte_stream_split_decoder.rs
index 9b2f43ace..b72ae8f62 100644
--- a/parquet/src/encodings/decoding/byte_stream_split_decoder.rs
+++ b/parquet/src/encodings/decoding/byte_stream_split_decoder.rs
@@ -21,7 +21,7 @@ use bytes::Bytes;
 
 use crate::basic::{Encoding, Type};
 use crate::data_type::private::ParquetValueType;
-use crate::data_type::{DataType, FixedLenByteArray, SliceAsBytes};
+use crate::data_type::{DataType, SliceAsBytes};
 use crate::errors::{ParquetError, Result};
 
 use super::Decoder;
@@ -234,12 +234,7 @@ impl<T: DataType> Decoder<T> for VariableWidthByteStreamSplitDecoder<T> {
         for (i, bi) in buffer.iter_mut().enumerate().take(num_values) {
             // Get a view into the data, without also copying the bytes
             let data = bytes_with_data.slice(i * type_size..(i + 1) * type_size);
-            // TODO: perhaps add a `set_from_bytes` method to `DataType` to avoid downcasting
-            let bi = bi
-                .as_mut_any()
-                .downcast_mut::<FixedLenByteArray>()
-                .expect("Decoding fixed length byte array");
-            bi.set_data(data);
+            bi.set_from_bytes(data);
         }
 
         Ok(num_values)
diff --git a/parquet/src/file/metadata/memory.rs b/parquet/src/file/metadata/memory.rs
index bb822b4cc..ad4522679 100644
--- a/parquet/src/file/metadata/memory.rs
+++ b/parquet/src/file/metadata/memory.rs
@@ -183,14 +183,8 @@ impl<T: ParquetValueType> HeapSize for PageIndex<T> {
 
 impl<T: ParquetValueType> HeapSize for ValueStatistics<T> {
     fn heap_size(&self) -> usize {
-        if self.has_min_max_set() {
-            return self.min().heap_size() + self.max().heap_size();
-        } else if self.min_is_exact() {
-            return self.min().heap_size();
-        } else if self.max_is_exact() {
-            return self.max().heap_size();
-        }
-        0
+        self.min_opt().map(T::heap_size).unwrap_or(0)
+            + self.max_opt().map(T::heap_size).unwrap_or(0)
     }
 }
 impl HeapSize for bool {
diff --git a/parquet/src/file/metadata/mod.rs b/parquet/src/file/metadata/mod.rs
index 45ef0c546..f3c73ee81 100644
--- a/parquet/src/file/metadata/mod.rs
+++ b/parquet/src/file/metadata/mod.rs
@@ -50,10 +50,9 @@
 //! [`decode_metadata`]: crate::file::footer::decode_metadata
 //!
 //! Writing:
-//! * Write `ParquetMetaData` to bytes in memory: Not yet supported (see [#6002])
+//! * Write `ParquetMetaData` to bytes in memory: [`ParquetMetaDataWriter`]
 //! * Writes `ParquetMetaData` to an async target: Not yet supported
 //!
-//! [#6002]: https://github.com/apache/arrow-rs/issues/6002
 //!
 //! # Metadata Encodings and Structures
 //!
@@ -95,6 +94,7 @@
 //!                         * Same name, different struct
 //! ```
 mod memory;
+mod writer;
 
 use std::ops::Range;
 use std::sync::Arc;
@@ -115,6 +115,8 @@ use crate::schema::types::{
     ColumnDescPtr, ColumnDescriptor, ColumnPath, SchemaDescPtr, SchemaDescriptor,
     Type as SchemaType,
 };
+pub use writer::ParquetMetaDataWriter;
+pub(crate) use writer::ThriftMetadataWriter;
 
 /// Page level statistics for each column chunk of each row group.
 ///
@@ -1613,7 +1615,7 @@ mod tests {
             .iter()
             .map(|column_descr| {
                 ColumnChunkMetaData::builder(column_descr.clone())
-                    .set_statistics(Statistics::new::<i32>(None, None, None, 0, false))
+                    .set_statistics(Statistics::new::<i32>(None, None, None, None, false))
                     .build()
             })
             .collect::<Result<Vec<_>>>()
@@ -1651,7 +1653,13 @@ mod tests {
             .iter()
             .map(|column_descr| {
                 ColumnChunkMetaData::builder(column_descr.clone())
-                    .set_statistics(Statistics::new::<i32>(Some(0), Some(100), None, 0, false))
+                    .set_statistics(Statistics::new::<i32>(
+                        Some(0),
+                        Some(100),
+                        None,
+                        None,
+                        false,
+                    ))
                     .build()
             })
             .collect::<Result<Vec<_>>>()
@@ -1665,7 +1673,7 @@ mod tests {
         let row_group_meta_with_stats = vec![row_group_meta_with_stats];
 
         let parquet_meta = ParquetMetaData::new(file_metadata.clone(), row_group_meta_with_stats);
-        let base_expected_size = 2280;
+        let base_expected_size = 2312;
 
         assert_eq!(parquet_meta.memory_size(), base_expected_size);
 
@@ -1693,7 +1701,7 @@ mod tests {
             ]]),
         );
 
-        let bigger_expected_size = 2784;
+        let bigger_expected_size = 2816;
         // more set fields means more memory usage
         assert!(bigger_expected_size > base_expected_size);
         assert_eq!(parquet_meta.memory_size(), bigger_expected_size);
diff --git a/parquet/src/file/metadata/writer.rs b/parquet/src/file/metadata/writer.rs
new file mode 100644
index 000000000..3573e3020
--- /dev/null
+++ b/parquet/src/file/metadata/writer.rs
@@ -0,0 +1,622 @@
+// 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 crate::errors::Result;
+use crate::file::metadata::{KeyValue, ParquetMetaData};
+use crate::file::page_index::index::Index;
+use crate::file::writer::TrackedWrite;
+use crate::file::PARQUET_MAGIC;
+use crate::format::{ColumnIndex, OffsetIndex, RowGroup};
+use crate::schema::types;
+use crate::schema::types::{SchemaDescPtr, SchemaDescriptor, TypePtr};
+use crate::thrift::TSerializable;
+use std::io::Write;
+use std::sync::Arc;
+use thrift::protocol::TCompactOutputProtocol;
+
+/// Writes `crate::file::metadata` structures to a thrift encoded byte stream
+///
+/// See [`ParquetMetaDataWriter`] for background and example.
+pub(crate) struct ThriftMetadataWriter<'a, W: Write> {
+    buf: &'a mut TrackedWrite<W>,
+    schema: &'a TypePtr,
+    schema_descr: &'a SchemaDescPtr,
+    row_groups: Vec<RowGroup>,
+    column_indexes: Option<&'a [Vec<Option<ColumnIndex>>]>,
+    offset_indexes: Option<&'a [Vec<Option<OffsetIndex>>]>,
+    key_value_metadata: Option<Vec<KeyValue>>,
+    created_by: Option<String>,
+    writer_version: i32,
+}
+
+impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
+    /// Serialize all the offset indexes to `self.buf`,
+    ///
+    /// Note: also updates the `ColumnChunk::offset_index_offset` and
+    /// `ColumnChunk::offset_index_length` to reflect the position and length
+    /// of the serialized offset indexes.
+    fn write_offset_indexes(&mut self, offset_indexes: &[Vec<Option<OffsetIndex>>]) -> Result<()> {
+        // iter row group
+        // iter each column
+        // 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 => {}
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Serialize all the column indexes to the `self.buf`
+    ///
+    /// Note: also updates the `ColumnChunk::column_index_offset` and
+    /// `ColumnChunk::column_index_length` to reflect the position and length
+    /// of the serialized column indexes.
+    fn write_column_indexes(&mut self, column_indexes: &[Vec<Option<ColumnIndex>>]) -> Result<()> {
+        // iter row group
+        // iter each column
+        // 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 => {}
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Assembles and writes the final metadata to self.buf
+    pub fn finish(mut self) -> Result<crate::format::FileMetaData> {
+        let num_rows = self.row_groups.iter().map(|x| x.num_rows).sum();
+
+        // Write column indexes and offset indexes
+        if let Some(column_indexes) = self.column_indexes {
+            self.write_column_indexes(column_indexes)?;
+        }
+        if let Some(offset_indexes) = self.offset_indexes {
+            self.write_offset_indexes(offset_indexes)?;
+        }
+
+        // We only include ColumnOrder for leaf nodes.
+        // Currently only supported ColumnOrder is TypeDefinedOrder so we set this
+        // for all leaf nodes.
+        // Even if the column has an undefined sort order, such as INTERVAL, this
+        // is still technically the defined TYPEORDER so it should still be set.
+        let column_orders = (0..self.schema_descr.num_columns())
+            .map(|_| crate::format::ColumnOrder::TYPEORDER(crate::format::TypeDefinedOrder {}))
+            .collect();
+        // This field is optional, perhaps in cases where no min/max fields are set
+        // in any Statistics or ColumnIndex object in the whole file.
+        // But for simplicity we always set this field.
+        let column_orders = Some(column_orders);
+
+        let file_metadata = crate::format::FileMetaData {
+            num_rows,
+            row_groups: self.row_groups,
+            key_value_metadata: self.key_value_metadata.clone(),
+            version: self.writer_version,
+            schema: types::to_thrift(self.schema.as_ref())?,
+            created_by: self.created_by.clone(),
+            column_orders,
+            encryption_algorithm: None,
+            footer_signing_key_metadata: None,
+        };
+
+        // Write file metadata
+        let start_pos = self.buf.bytes_written();
+        {
+            let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
+            file_metadata.write_to_out_protocol(&mut protocol)?;
+        }
+        let end_pos = self.buf.bytes_written();
+
+        // Write footer
+        let metadata_len = (end_pos - start_pos) as u32;
+
+        self.buf.write_all(&metadata_len.to_le_bytes())?;
+        self.buf.write_all(&PARQUET_MAGIC)?;
+        Ok(file_metadata)
+    }
+
+    pub fn new(
+        buf: &'a mut TrackedWrite<W>,
+        schema: &'a TypePtr,
+        schema_descr: &'a SchemaDescPtr,
+        row_groups: Vec<RowGroup>,
+        created_by: Option<String>,
+        writer_version: i32,
+    ) -> Self {
+        Self {
+            buf,
+            schema,
+            schema_descr,
+            row_groups,
+            column_indexes: None,
+            offset_indexes: None,
+            key_value_metadata: None,
+            created_by,
+            writer_version,
+        }
+    }
+
+    pub fn with_column_indexes(mut self, column_indexes: &'a [Vec<Option<ColumnIndex>>]) -> Self {
+        self.column_indexes = Some(column_indexes);
+        self
+    }
+
+    pub fn with_offset_indexes(mut self, offset_indexes: &'a [Vec<Option<OffsetIndex>>]) -> Self {
+        self.offset_indexes = Some(offset_indexes);
+        self
+    }
+
+    pub fn with_key_value_metadata(mut self, key_value_metadata: Vec<KeyValue>) -> Self {
+        self.key_value_metadata = Some(key_value_metadata);
+        self
+    }
+}
+
+/// Writes [`ParquetMetaData`] to a byte stream
+///
+/// This structure handles the details of writing the various parts of Parquet
+/// metadata into a byte stream. It is used to write the metadata into a parquet
+/// file and can also write metadata into other locations (such as a store of
+/// bytes).
+///
+/// # Discussion
+///
+/// The process of writing Parquet metadata is tricky because the
+/// metadata is not stored as a single inline thrift structure. It can have
+/// several "out of band" structures such as the [`OffsetIndex`] and
+/// BloomFilters stored in separate structures whose locations are stored as
+/// offsets from the beginning of the file.
+///
+/// Note: this writer does not directly write BloomFilters. In order to write
+/// BloomFilters, write the bloom filters into the buffer before creating the
+/// metadata writer. Then set the corresponding `bloom_filter_offset` and
+/// `bloom_filter_length` on [`ColumnChunkMetaData`] passed to this writer.
+///
+/// # Output Format
+///
+/// The format of the metadata is as follows:
+///
+/// 1. Optional [`ColumnIndex`] (thrift encoded)
+/// 2. Optional [`OffsetIndex`] (thrift encoded)
+/// 3. [`FileMetaData`] (thrift encoded)
+/// 4. Length of encoded `FileMetaData` (4 bytes, little endian)
+/// 5. Parquet Magic Bytes (4 bytes)
+///
+/// [`FileMetaData`]: crate::format::FileMetaData
+/// [`ColumnChunkMetaData`]: crate::file::metadata::ColumnChunkMetaData
+///
+/// ```text
+/// ┌──────────────────────┐
+/// │                      │
+/// │         ...          │
+/// │                      │
+/// │┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ │
+/// │     ColumnIndex     ◀│─ ─ ─
+/// ││    (Optional)     │ │     │
+/// │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─  │
+/// │┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ │     │ FileMetadata
+/// │     OffsetIndex      │       contains embedded
+/// ││    (Optional)     │◀┼ ─   │ offsets to
+/// │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─  │  │    ColumnIndex and
+/// │╔═══════════════════╗ │     │ OffsetIndex
+/// │║                   ║ │  │
+/// │║                   ║ ┼ ─   │
+/// │║   FileMetadata    ║ │
+/// │║                   ║ ┼ ─ ─ ┘
+/// │║                   ║ │
+/// │╚═══════════════════╝ │
+/// │┌───────────────────┐ │
+/// ││  metadata length  │ │ length of FileMetadata  (only)
+/// │└───────────────────┘ │
+/// │┌───────────────────┐ │
+/// ││      'PAR1'       │ │ Parquet Magic Bytes
+/// │└───────────────────┘ │
+/// └──────────────────────┘
+///      Output Buffer
+/// ```
+///
+/// # Example
+/// ```no_run
+/// # use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataWriter};
+/// # fn get_metadata() -> ParquetMetaData { unimplemented!(); }
+/// // write parquet metadata to an in-memory buffer
+/// let mut buffer = vec![];
+/// let metadata: ParquetMetaData = get_metadata();
+/// let writer = ParquetMetaDataWriter::new(&mut buffer, &metadata);
+/// // write the metadata to the buffer
+/// writer.finish().unwrap();
+/// assert!(!buffer.is_empty());
+/// ```
+pub struct ParquetMetaDataWriter<'a, W: Write> {
+    buf: TrackedWrite<W>,
+    metadata: &'a ParquetMetaData,
+}
+
+impl<'a, W: Write> ParquetMetaDataWriter<'a, W> {
+    /// Create a new `ParquetMetaDataWriter` to write to `buf`
+    ///
+    /// Note any embedded offsets in the metadata will be written assuming the
+    /// metadata is at the start of the buffer. If the metadata is being written
+    /// to a location other than the start of the buffer, see [`Self::new_with_tracked`]
+    ///
+    /// See example on the struct level documentation
+    pub fn new(buf: W, metadata: &'a ParquetMetaData) -> Self {
+        Self::new_with_tracked(TrackedWrite::new(buf), metadata)
+    }
+
+    /// Create a new ParquetMetaDataWriter to write to `buf`
+    ///
+    /// This method is used when the metadata is being written to a location other
+    /// than the start of the buffer.
+    ///
+    /// See example on the struct level documentation
+    pub fn new_with_tracked(buf: TrackedWrite<W>, metadata: &'a ParquetMetaData) -> Self {
+        Self { buf, metadata }
+    }
+
+    pub fn finish(mut self) -> Result<()> {
+        let file_metadata = self.metadata.file_metadata();
+
+        let schema = Arc::new(file_metadata.schema().clone());
+        let schema_descr = Arc::new(SchemaDescriptor::new(schema.clone()));
+        let created_by = file_metadata.created_by().map(str::to_string);
+
+        let row_groups = self
+            .metadata
+            .row_groups()
+            .iter()
+            .map(|rg| rg.to_thrift())
+            .collect::<Vec<_>>();
+
+        let key_value_metadata = file_metadata.key_value_metadata().cloned();
+
+        let column_indexes = self.convert_column_indexes();
+        let offset_indexes = self.convert_offset_index();
+
+        let mut encoder = ThriftMetadataWriter::new(
+            &mut self.buf,
+            &schema,
+            &schema_descr,
+            row_groups,
+            created_by,
+            file_metadata.version(),
+        );
+        encoder = encoder.with_column_indexes(&column_indexes);
+        encoder = encoder.with_offset_indexes(&offset_indexes);
+        if let Some(key_value_metadata) = key_value_metadata {
+            encoder = encoder.with_key_value_metadata(key_value_metadata);
+        }
+        encoder.finish()?;
+
+        Ok(())
+    }
+
+    fn convert_column_indexes(&self) -> Vec<Vec<Option<ColumnIndex>>> {
+        if let Some(row_group_column_indexes) = self.metadata.column_index() {
+            (0..self.metadata.row_groups().len())
+                .map(|rg_idx| {
+                    let column_indexes = &row_group_column_indexes[rg_idx];
+                    column_indexes
+                        .iter()
+                        .map(|column_index| match column_index {
+                            Index::NONE => None,
+                            Index::BOOLEAN(column_index) => Some(column_index.to_thrift()),
+                            Index::BYTE_ARRAY(column_index) => Some(column_index.to_thrift()),
+                            Index::DOUBLE(column_index) => Some(column_index.to_thrift()),
+                            Index::FIXED_LEN_BYTE_ARRAY(column_index) => {
+                                Some(column_index.to_thrift())
+                            }
+                            Index::FLOAT(column_index) => Some(column_index.to_thrift()),
+                            Index::INT32(column_index) => Some(column_index.to_thrift()),
+                            Index::INT64(column_index) => Some(column_index.to_thrift()),
+                            Index::INT96(column_index) => Some(column_index.to_thrift()),
+                        })
+                        .collect()
+                })
+                .collect()
+        } else {
+            // make a None for each row group, for each column
+            self.metadata
+                .row_groups()
+                .iter()
+                .map(|rg| std::iter::repeat(None).take(rg.columns().len()).collect())
+                .collect()
+        }
+    }
+
+    fn convert_offset_index(&self) -> Vec<Vec<Option<OffsetIndex>>> {
+        if let Some(row_group_offset_indexes) = self.metadata.offset_index() {
+            (0..self.metadata.row_groups().len())
+                .map(|rg_idx| {
+                    let offset_indexes = &row_group_offset_indexes[rg_idx];
+                    offset_indexes
+                        .iter()
+                        .map(|offset_index| Some(offset_index.to_thrift()))
+                        .collect()
+                })
+                .collect()
+        } else {
+            // make a None for each row group, for each column
+            self.metadata
+                .row_groups()
+                .iter()
+                .map(|rg| std::iter::repeat(None).take(rg.columns().len()).collect())
+                .collect()
+        }
+    }
+}
+
+#[cfg(test)]
+#[cfg(feature = "arrow")]
+#[cfg(feature = "async")]
+mod tests {
+    use std::sync::Arc;
+
+    use crate::file::footer::parse_metadata;
+    use crate::file::metadata::{
+        ColumnChunkMetaData, ParquetMetaData, ParquetMetaDataWriter, RowGroupMetaData,
+    };
+    use crate::file::properties::{EnabledStatistics, WriterProperties};
+    use crate::file::reader::{FileReader, SerializedFileReader};
+    use crate::{
+        arrow::ArrowWriter,
+        file::{page_index::index::Index, serialized_reader::ReadOptionsBuilder},
+    };
+    use arrow_array::{ArrayRef, Int32Array, RecordBatch};
+    use arrow_schema::{DataType as ArrowDataType, Field, Schema};
+    use bytes::{BufMut, Bytes, BytesMut};
+
+    struct TestMetadata {
+        #[allow(dead_code)]
+        file_size: usize,
+        metadata: ParquetMetaData,
+    }
+
+    fn has_page_index(metadata: &ParquetMetaData) -> bool {
+        match metadata.column_index() {
+            Some(column_index) => column_index
+                .iter()
+                .any(|rg_idx| rg_idx.iter().all(|col_idx| !matches!(col_idx, Index::NONE))),
+            None => false,
+        }
+    }
+
+    #[test]
+    fn test_roundtrip_parquet_metadata_without_page_index() {
+        // We currently don't have an ad-hoc ParquetMetadata loader that can load page indexes so
+        // we at least test round trip without them
+        let metadata = get_test_metadata(false, false);
+        assert!(!has_page_index(&metadata.metadata));
+
+        let mut buf = BytesMut::new().writer();
+        {
+            let writer = ParquetMetaDataWriter::new(&mut buf, &metadata.metadata);
+            writer.finish().unwrap();
+        }
+
+        let data = buf.into_inner().freeze();
+
+        let decoded_metadata = parse_metadata(&data).unwrap();
+        assert!(!has_page_index(&metadata.metadata));
+
+        assert_eq!(metadata.metadata, decoded_metadata);
+    }
+
+    fn get_test_metadata(write_page_index: bool, read_page_index: bool) -> TestMetadata {
+        let mut buf = BytesMut::new().writer();
+        let schema: Arc<Schema> = Arc::new(Schema::new(vec![Field::new(
+            "a",
+            ArrowDataType::Int32,
+            true,
+        )]));
+
+        let a: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None, Some(2)]));
+
+        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 mut writer = ArrowWriter::try_new(&mut buf, schema, Some(writer_props)).unwrap();
+        writer.write(&batch).unwrap();
+        writer.close().unwrap();
+
+        let data = buf.into_inner().freeze();
+
+        let reader_opts = match read_page_index {
+            true => ReadOptionsBuilder::new().with_page_index().build(),
+            false => ReadOptionsBuilder::new().build(),
+        };
+        let reader = SerializedFileReader::new_with_options(data.clone(), reader_opts).unwrap();
+        let metadata = reader.metadata().clone();
+        TestMetadata {
+            file_size: data.len(),
+            metadata,
+        }
+    }
+
+    /// Temporary function so we can test loading metadata with page indexes
+    /// while we haven't fully figured out how to load it cleanly
+    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;
+
+        /// Adapt a `Bytes` to a `MetadataFetch` implementation.
+        struct AsyncBytes {
+            data: Bytes,
+        }
+
+        impl AsyncBytes {
+            fn new(data: Bytes) -> Self {
+                Self { data }
+            }
+        }
+
+        impl MetadataFetch for AsyncBytes {
+            fn fetch(&mut self, range: Range<usize>) -> BoxFuture<'_, ParquetResult<Bytes>> {
+                async move { Ok(self.data.slice(range.start..range.end)) }.boxed()
+            }
+        }
+
+        /// A `MetadataFetch` implementation that reads from a subset of the full data
+        /// while accepting ranges that address the full data.
+        struct MaskedBytes {
+            inner: Box<dyn MetadataFetch + Send>,
+            inner_range: Range<usize>,
+        }
+
+        impl MaskedBytes {
+            fn new(inner: Box<dyn MetadataFetch + Send>, inner_range: Range<usize>) -> Self {
+                Self { inner, inner_range }
+            }
+        }
+
+        impl MetadataFetch for &mut MaskedBytes {
+            fn fetch(&mut self, range: Range<usize>) -> BoxFuture<'_, ParquetResult<Bytes>> {
+                let inner_range = self.inner_range.clone();
+                println!("inner_range: {:?}", inner_range);
+                println!("range: {:?}", range);
+                assert!(inner_range.start <= range.start && inner_range.end >= range.end);
+                let range =
+                    range.start - self.inner_range.start..range.end - self.inner_range.start;
+                self.inner.fetch(range)
+            }
+        }
+
+        let metadata_length = data.len();
+        let mut reader = MaskedBytes::new(
+            Box::new(AsyncBytes::new(data)),
+            file_size - metadata_length..file_size,
+        );
+        let metadata = MetadataLoader::load(&mut reader, file_size, None)
+            .await
+            .unwrap();
+        let loaded_metadata = metadata.finish();
+        let mut metadata = MetadataLoader::new(&mut reader, loaded_metadata);
+        metadata.load_page_index(true, true).await.unwrap();
+        metadata.finish()
+    }
+
+    fn check_columns_are_equivalent(left: &ColumnChunkMetaData, right: &ColumnChunkMetaData) {
+        assert_eq!(left.column_descr(), right.column_descr());
+        assert_eq!(left.encodings(), right.encodings());
+        assert_eq!(left.num_values(), right.num_values());
+        assert_eq!(left.compressed_size(), right.compressed_size());
+        assert_eq!(left.data_page_offset(), right.data_page_offset());
+        assert_eq!(left.statistics(), right.statistics());
+        assert_eq!(left.offset_index_length(), right.offset_index_length());
+        assert_eq!(left.column_index_length(), right.column_index_length());
+        assert_eq!(
+            left.unencoded_byte_array_data_bytes(),
+            right.unencoded_byte_array_data_bytes()
+        );
+    }
+
+    fn check_row_groups_are_equivalent(left: &RowGroupMetaData, right: &RowGroupMetaData) {
+        assert_eq!(left.num_rows(), right.num_rows());
+        assert_eq!(left.file_offset(), right.file_offset());
+        assert_eq!(left.total_byte_size(), right.total_byte_size());
+        assert_eq!(left.schema_descr(), right.schema_descr());
+        assert_eq!(left.num_columns(), right.num_columns());
+        left.columns()
+            .iter()
+            .zip(right.columns().iter())
+            .for_each(|(lc, rc)| {
+                check_columns_are_equivalent(lc, rc);
+            });
+    }
+
+    #[tokio::test]
+    async fn test_encode_parquet_metadata_with_page_index() {
+        // Create a ParquetMetadata with page index information
+        let metadata = get_test_metadata(true, true);
+        assert!(has_page_index(&metadata.metadata));
+
+        let mut buf = BytesMut::new().writer();
+        {
+            let writer = ParquetMetaDataWriter::new(&mut buf, &metadata.metadata);
+            writer.finish().unwrap();
+        }
+
+        let data = buf.into_inner().freeze();
+
+        let decoded_metadata = load_metadata_from_bytes(data.len(), data).await;
+
+        // Because the page index offsets will differ, compare invariant parts of the metadata
+        assert_eq!(
+            metadata.metadata.file_metadata(),
+            decoded_metadata.file_metadata()
+        );
+        assert_eq!(
+            metadata.metadata.column_index(),
+            decoded_metadata.column_index()
+        );
+        assert_eq!(
+            metadata.metadata.offset_index(),
+            decoded_metadata.offset_index()
+        );
+        assert_eq!(
+            metadata.metadata.num_row_groups(),
+            decoded_metadata.num_row_groups()
+        );
+
+        metadata
+            .metadata
+            .row_groups()
+            .iter()
+            .zip(decoded_metadata.row_groups().iter())
+            .for_each(|(left, right)| {
+                check_row_groups_are_equivalent(left, right);
+            });
+    }
+}
diff --git a/parquet/src/file/serialized_reader.rs b/parquet/src/file/serialized_reader.rs
index 07e80f7f6..0a3e51931 100644
--- a/parquet/src/file/serialized_reader.rs
+++ b/parquet/src/file/serialized_reader.rs
@@ -1544,7 +1544,10 @@ mod tests {
 
     fn get_row_group_min_max_bytes(r: &RowGroupMetaData, col_num: usize) -> (&[u8], &[u8]) {
         let statistics = r.column(col_num).statistics().unwrap();
-        (statistics.min_bytes(), statistics.max_bytes())
+        (
+            statistics.min_bytes_opt().unwrap_or_default(),
+            statistics.max_bytes_opt().unwrap_or_default(),
+        )
     }
 
     #[test]
diff --git a/parquet/src/file/statistics.rs b/parquet/src/file/statistics.rs
index dded9c9c1..4134685ff 100644
--- a/parquet/src/file/statistics.rs
+++ b/parquet/src/file/statistics.rs
@@ -24,17 +24,16 @@
 //! ```rust
 //! use parquet::file::statistics::Statistics;
 //!
-//! let stats = Statistics::int32(Some(1), Some(10), None, 3, true);
-//! assert_eq!(stats.null_count(), 3);
-//! assert!(stats.has_min_max_set());
+//! let stats = Statistics::int32(Some(1), Some(10), None, Some(3), true);
+//! assert_eq!(stats.null_count_opt(), Some(3));
 //! assert!(stats.is_min_max_deprecated());
 //! assert!(stats.min_is_exact());
 //! assert!(stats.max_is_exact());
 //!
 //! match stats {
 //!     Statistics::Int32(ref typed) => {
-//!         assert_eq!(*typed.min(), 1);
-//!         assert_eq!(*typed.max(), 10);
+//!         assert_eq!(typed.min_opt(), Some(&1));
+//!         assert_eq!(typed.max_opt(), Some(&10));
 //!     }
 //!     _ => {}
 //! }
@@ -89,7 +88,7 @@ macro_rules! statistics_new_func {
             min: $vtype,
             max: $vtype,
             distinct: Option<u64>,
-            nulls: u64,
+            nulls: Option<u64>,
             is_deprecated: bool,
         ) -> Self {
             Statistics::$stat(ValueStatistics::new(
@@ -127,6 +126,8 @@ pub fn from_thrift(
     Ok(match thrift_stats {
         Some(stats) => {
             // Number of nulls recorded, when it is not available, we just mark it as 0.
+            // TODO this should be `None` if there is no information about NULLS.
+            // see https://github.com/apache/arrow-rs/pull/6216/files
             let null_count = stats.null_count.unwrap_or(0);
 
             if null_count < 0 {
@@ -137,7 +138,7 @@ pub fn from_thrift(
             }
 
             // Generic null count.
-            let null_count = null_count as u64;
+            let null_count = Some(null_count as u64);
             // Generic distinct count (count of distinct values occurring)
             let distinct_count = stats.distinct_count.map(|value| value as u64);
             // Whether or not statistics use deprecated min/max fields.
@@ -243,14 +244,19 @@ pub fn from_thrift(
 pub fn to_thrift(stats: Option<&Statistics>) -> Option<TStatistics> {
     let stats = stats?;
 
+    // record null counts if greater than zero.
+    //
+    // TODO: This should be Some(0) if there are no nulls.
+    // see https://github.com/apache/arrow-rs/pull/6216/files
+    let null_count = stats
+        .null_count_opt()
+        .map(|value| value as i64)
+        .filter(|&x| x > 0);
+
     let mut thrift_stats = TStatistics {
         max: None,
         min: None,
-        null_count: if stats.has_nulls() {
-            Some(stats.null_count() as i64)
-        } else {
-            None
-        },
+        null_count,
         distinct_count: stats.distinct_count().map(|value| value as i64),
         max_value: None,
         min_value: None,
@@ -259,17 +265,12 @@ pub fn to_thrift(stats: Option<&Statistics>) -> Option<TStatistics> {
     };
 
     // Get min/max if set.
-    let (min, max, min_exact, max_exact) = if stats.has_min_max_set() {
-        (
-            Some(stats.min_bytes().to_vec()),
-            Some(stats.max_bytes().to_vec()),
-            Some(stats.min_is_exact()),
-            Some(stats.max_is_exact()),
-        )
-    } else {
-        (None, None, None, None)
-    };
-
+    let (min, max, min_exact, max_exact) = (
+        stats.min_bytes_opt().map(|x| x.to_vec()),
+        stats.max_bytes_opt().map(|x| x.to_vec()),
+        Some(stats.min_is_exact()),
+        Some(stats.max_is_exact()),
+    );
     if stats.is_min_max_backwards_compatible() {
         // Copy to deprecated min, max values for compatibility with older readers
         thrift_stats.min.clone_from(&min);
@@ -321,7 +322,7 @@ impl Statistics {
         min: Option<T>,
         max: Option<T>,
         distinct_count: Option<u64>,
-        null_count: u64,
+        null_count: Option<u64>,
         is_deprecated: bool,
     ) -> Self {
         Self::from(ValueStatistics::new(
@@ -385,19 +386,39 @@ impl Statistics {
 
     /// Returns number of null values for the column.
     /// Note that this includes all nulls when column is part of the complex type.
+    ///
+    /// Note this API returns 0 if the null count is not available.
+    #[deprecated(since = "53.0.0", note = "Use `null_count_opt` method instead")]
     pub fn null_count(&self) -> u64 {
-        statistics_enum_func![self, null_count]
+        // 0 to remain consistent behavior prior to `null_count_opt`
+        self.null_count_opt().unwrap_or(0)
     }
 
     /// Returns `true` if statistics collected any null values, `false` otherwise.
+    #[deprecated(since = "53.0.0", note = "Use `null_count_opt` method instead")]
+    #[allow(deprecated)]
     pub fn has_nulls(&self) -> bool {
         self.null_count() > 0
     }
 
-    /// Returns `true` if min value and max value are set.
+    /// Returns number of null values for the column, if known.
+    /// Note that this includes all nulls when column is part of the complex type.
+    ///
+    /// Note this API returns Some(0) even if the null count was not present
+    /// in the statistics.
+    /// See <https://github.com/apache/arrow-rs/pull/6216/files>
+    pub fn null_count_opt(&self) -> Option<u64> {
+        statistics_enum_func![self, null_count_opt]
+    }
+
+    /// Whether or not min and max values are set.
     /// Normally both min/max values will be set to `Some(value)` or `None`.
+    #[deprecated(
+        since = "53.0.0",
+        note = "Use `min_bytes_opt` and `max_bytes_opt` methods instead"
+    )]
     pub fn has_min_max_set(&self) -> bool {
-        statistics_enum_func![self, has_min_max_set]
+        statistics_enum_func![self, _internal_has_min_max_set]
     }
 
     /// Returns `true` if the min value is set, and is an exact min value.
@@ -410,16 +431,28 @@ impl Statistics {
         statistics_enum_func![self, max_is_exact]
     }
 
+    /// Returns slice of bytes that represent min value, if min value is known.
+    pub fn min_bytes_opt(&self) -> Option<&[u8]> {
+        statistics_enum_func![self, min_bytes_opt]
+    }
+
     /// Returns slice of bytes that represent min value.
     /// Panics if min value is not set.
+    #[deprecated(since = "53.0.0", note = "Use `max_bytes_opt` instead")]
     pub fn min_bytes(&self) -> &[u8] {
-        statistics_enum_func![self, min_bytes]
+        self.min_bytes_opt().unwrap()
+    }
+
+    /// Returns slice of bytes that represent max value, if max value is known.
+    pub fn max_bytes_opt(&self) -> Option<&[u8]> {
+        statistics_enum_func![self, max_bytes_opt]
     }
 
     /// Returns slice of bytes that represent max value.
     /// Panics if max value is not set.
+    #[deprecated(since = "53.0.0", note = "Use `max_bytes_opt` instead")]
     pub fn max_bytes(&self) -> &[u8] {
-        statistics_enum_func![self, max_bytes]
+        self.max_bytes_opt().unwrap()
     }
 
     /// Returns physical type associated with statistics.
@@ -464,7 +497,7 @@ pub struct ValueStatistics<T> {
     max: Option<T>,
     // Distinct count could be omitted in some cases
     distinct_count: Option<u64>,
-    null_count: u64,
+    null_count: Option<u64>,
 
     // Whether or not the min or max values are exact, or truncated.
     is_max_value_exact: bool,
@@ -485,7 +518,7 @@ impl<T: ParquetValueType> ValueStatistics<T> {
         min: Option<T>,
         max: Option<T>,
         distinct_count: Option<u64>,
-        null_count: u64,
+        null_count: Option<u64>,
         is_min_max_deprecated: bool,
     ) -> Self {
         Self {
@@ -538,37 +571,68 @@ impl<T: ParquetValueType> ValueStatistics<T> {
     ///
     /// Panics if min value is not set, e.g. all values are `null`.
     /// Use `has_min_max_set` method to check that.
+    #[deprecated(since = "53.0.0", note = "Use `min_opt` instead")]
     pub fn min(&self) -> &T {
         self.min.as_ref().unwrap()
     }
 
+    /// Returns min value of the statistics, if known.
+    pub fn min_opt(&self) -> Option<&T> {
+        self.min.as_ref()
+    }
+
     /// Returns max value of the statistics.
     ///
     /// Panics if max value is not set, e.g. all values are `null`.
     /// Use `has_min_max_set` method to check that.
+    #[deprecated(since = "53.0.0", note = "Use `max_opt` instead")]
     pub fn max(&self) -> &T {
         self.max.as_ref().unwrap()
     }
 
+    /// Returns max value of the statistics, if known.
+    pub fn max_opt(&self) -> Option<&T> {
+        self.max.as_ref()
+    }
+
+    /// Returns min value as bytes of the statistics, if min value is known.
+    pub fn min_bytes_opt(&self) -> Option<&[u8]> {
+        self.min_opt().map(AsBytes::as_bytes)
+    }
+
     /// Returns min value as bytes of the statistics.
     ///
     /// Panics if min value is not set, use `has_min_max_set` method to check
     /// if values are set.
+    #[deprecated(since = "53.0.0", note = "Use `min_bytes_opt` instead")]
     pub fn min_bytes(&self) -> &[u8] {
-        self.min().as_bytes()
+        self.min_bytes_opt().unwrap()
+    }
+
+    /// Returns max value as bytes of the statistics, if max value is known.
+    pub fn max_bytes_opt(&self) -> Option<&[u8]> {
+        self.max_opt().map(AsBytes::as_bytes)
     }
 
     /// Returns max value as bytes of the statistics.
     ///
     /// Panics if max value is not set, use `has_min_max_set` method to check
     /// if values are set.
+    #[deprecated(since = "53.0.0", note = "Use `max_bytes_opt` instead")]
     pub fn max_bytes(&self) -> &[u8] {
-        self.max().as_bytes()
+        self.max_bytes_opt().unwrap()
     }
 
     /// Whether or not min and max values are set.
     /// Normally both min/max values will be set to `Some(value)` or `None`.
+    #[deprecated(since = "53.0.0", note = "Use `min_opt` and `max_opt` methods instead")]
     pub fn has_min_max_set(&self) -> bool {
+        self._internal_has_min_max_set()
+    }
+
+    /// Whether or not min and max values are set.
+    /// Normally both min/max values will be set to `Some(value)` or `None`.
+    pub(crate) fn _internal_has_min_max_set(&self) -> bool {
         self.min.is_some() && self.max.is_some()
     }
 
@@ -587,8 +651,16 @@ impl<T: ParquetValueType> ValueStatistics<T> {
         self.distinct_count
     }
 
-    /// Returns null count.
+    /// Returns number of null values for the column.
+    /// Note that this includes all nulls when column is part of the complex type.
+    #[deprecated(since = "53.0.0", note = "Use `null_count_opt` method instead")]
     pub fn null_count(&self) -> u64 {
+        // 0 to remain consistent behavior prior to `null_count_opt`
+        self.null_count_opt().unwrap_or(0)
+    }
+
+    /// Returns null count.
+    pub fn null_count_opt(&self) -> Option<u64> {
         self.null_count
     }
 
@@ -630,7 +702,11 @@ impl<T: ParquetValueType> fmt::Display for ValueStatistics<T> {
             Some(value) => write!(f, "{value}")?,
             None => write!(f, "N/A")?,
         }
-        write!(f, ", null_count: {}", self.null_count)?;
+        write!(f, ", null_count: ")?;
+        match self.null_count {
+            Some(value) => write!(f, "{value}")?,
+            None => write!(f, "N/A")?,
+        }
         write!(f, ", min_max_deprecated: {}", self.is_min_max_deprecated)?;
         write!(f, ", max_value_exact: {}", self.is_max_value_exact)?;
         write!(f, ", min_value_exact: {}", self.is_min_value_exact)?;
@@ -642,7 +718,7 @@ impl<T: ParquetValueType> fmt::Debug for ValueStatistics<T> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(
             f,
-            "{{min: {:?}, max: {:?}, distinct_count: {:?}, null_count: {}, \
+            "{{min: {:?}, max: {:?}, distinct_count: {:?}, null_count: {:?}, \
              min_max_deprecated: {}, min_max_backwards_compatible: {}, max_value_exact: {}, min_value_exact: {}}}",
             self.min,
             self.max,
@@ -662,21 +738,19 @@ mod tests {
 
     #[test]
     fn test_statistics_min_max_bytes() {
-        let stats = Statistics::int32(Some(-123), Some(234), None, 1, false);
-        assert!(stats.has_min_max_set());
-        assert_eq!(stats.min_bytes(), (-123).as_bytes());
-        assert_eq!(stats.max_bytes(), 234.as_bytes());
+        let stats = Statistics::int32(Some(-123), Some(234), None, Some(1), false);
+        assert_eq!(stats.min_bytes_opt(), Some((-123).as_bytes()));
+        assert_eq!(stats.max_bytes_opt(), Some(234.as_bytes()));
 
         let stats = Statistics::byte_array(
             Some(ByteArray::from(vec![1, 2, 3])),
             Some(ByteArray::from(vec![3, 4, 5])),
             None,
-            1,
+            Some(1),
             true,
         );
-        assert!(stats.has_min_max_set());
-        assert_eq!(stats.min_bytes(), &[1, 2, 3]);
-        assert_eq!(stats.max_bytes(), &[3, 4, 5]);
+        assert_eq!(stats.min_bytes_opt().unwrap(), &[1, 2, 3]);
+        assert_eq!(stats.max_bytes_opt().unwrap(), &[3, 4, 5]);
     }
 
     #[test]
@@ -704,30 +778,30 @@ mod tests {
 
     #[test]
     fn test_statistics_debug() {
-        let stats = Statistics::int32(Some(1), Some(12), None, 12, true);
+        let stats = Statistics::int32(Some(1), Some(12), None, Some(12), true);
         assert_eq!(
             format!("{stats:?}"),
-            "Int32({min: Some(1), max: Some(12), distinct_count: None, null_count: 12, \
+            "Int32({min: Some(1), max: Some(12), distinct_count: None, null_count: Some(12), \
              min_max_deprecated: true, min_max_backwards_compatible: true, max_value_exact: true, min_value_exact: true})"
         );
 
-        let stats = Statistics::int32(None, None, None, 7, false);
+        let stats = Statistics::int32(None, None, None, Some(7), false);
         assert_eq!(
             format!("{stats:?}"),
-            "Int32({min: None, max: None, distinct_count: None, null_count: 7, \
+            "Int32({min: None, max: None, distinct_count: None, null_count: Some(7), \
              min_max_deprecated: false, min_max_backwards_compatible: false, max_value_exact: false, min_value_exact: false})"
         )
     }
 
     #[test]
     fn test_statistics_display() {
-        let stats = Statistics::int32(Some(1), Some(12), None, 12, true);
+        let stats = Statistics::int32(Some(1), Some(12), None, Some(12), true);
         assert_eq!(
             format!("{stats}"),
             "{min: 1, max: 12, distinct_count: N/A, null_count: 12, min_max_deprecated: true, max_value_exact: true, min_value_exact: true}"
         );
 
-        let stats = Statistics::int64(None, None, None, 7, false);
+        let stats = Statistics::int64(None, None, None, Some(7), false);
         assert_eq!(
             format!("{stats}"),
             "{min: N/A, max: N/A, distinct_count: N/A, null_count: 7, min_max_deprecated: \
@@ -738,7 +812,7 @@ mod tests {
             Some(Int96::from(vec![1, 0, 0])),
             Some(Int96::from(vec![2, 3, 4])),
             None,
-            3,
+            Some(3),
             true,
         );
         assert_eq!(
@@ -752,7 +826,7 @@ mod tests {
                 Some(ByteArray::from(vec![1u8])),
                 Some(ByteArray::from(vec![2u8])),
                 Some(5),
-                7,
+                Some(7),
                 false,
             )
             .with_max_is_exact(false)
@@ -766,22 +840,22 @@ mod tests {
 
     #[test]
     fn test_statistics_partial_eq() {
-        let expected = Statistics::int32(Some(12), Some(45), None, 11, true);
+        let expected = Statistics::int32(Some(12), Some(45), None, Some(11), true);
 
-        assert!(Statistics::int32(Some(12), Some(45), None, 11, true) == expected);
-        assert!(Statistics::int32(Some(11), Some(45), None, 11, true) != expected);
-        assert!(Statistics::int32(Some(12), Some(44), None, 11, true) != expected);
-        assert!(Statistics::int32(Some(12), Some(45), None, 23, true) != expected);
-        assert!(Statistics::int32(Some(12), Some(45), None, 11, false) != expected);
+        assert!(Statistics::int32(Some(12), Some(45), None, Some(11), true) == expected);
+        assert!(Statistics::int32(Some(11), Some(45), None, Some(11), true) != expected);
+        assert!(Statistics::int32(Some(12), Some(44), None, Some(11), true) != expected);
+        assert!(Statistics::int32(Some(12), Some(45), None, Some(23), true) != expected);
+        assert!(Statistics::int32(Some(12), Some(45), None, Some(11), false) != expected);
 
         assert!(
-            Statistics::int32(Some(12), Some(45), None, 11, false)
-                != Statistics::int64(Some(12), Some(45), None, 11, false)
+            Statistics::int32(Some(12), Some(45), None, Some(11), false)
+                != Statistics::int64(Some(12), Some(45), None, Some(11), false)
         );
 
         assert!(
-            Statistics::boolean(Some(false), Some(true), None, 0, true)
-                != Statistics::double(Some(1.2), Some(4.5), None, 0, true)
+            Statistics::boolean(Some(false), Some(true), None, None, true)
+                != Statistics::double(Some(1.2), Some(4.5), None, None, true)
         );
 
         assert!(
@@ -789,13 +863,13 @@ mod tests {
                 Some(ByteArray::from(vec![1, 2, 3])),
                 Some(ByteArray::from(vec![1, 2, 3])),
                 None,
-                0,
+                None,
                 true
             ) != Statistics::fixed_len_byte_array(
                 Some(ByteArray::from(vec![1, 2, 3]).into()),
                 Some(ByteArray::from(vec![1, 2, 3]).into()),
                 None,
-                0,
+                None,
                 true,
             )
         );
@@ -805,14 +879,14 @@ mod tests {
                 Some(ByteArray::from(vec![1, 2, 3])),
                 Some(ByteArray::from(vec![1, 2, 3])),
                 None,
-                0,
+                None,
                 true,
             ) != Statistics::ByteArray(
                 ValueStatistics::new(
                     Some(ByteArray::from(vec![1, 2, 3])),
                     Some(ByteArray::from(vec![1, 2, 3])),
                     None,
-                    0,
+                    None,
                     true,
                 )
                 .with_max_is_exact(false)
@@ -824,14 +898,14 @@ mod tests {
                 Some(FixedLenByteArray::from(vec![1, 2, 3])),
                 Some(FixedLenByteArray::from(vec![1, 2, 3])),
                 None,
-                0,
+                None,
                 true,
             ) != Statistics::FixedLenByteArray(
                 ValueStatistics::new(
                     Some(FixedLenByteArray::from(vec![1, 2, 3])),
                     Some(FixedLenByteArray::from(vec![1, 2, 3])),
                     None,
-                    0,
+                    None,
                     true,
                 )
                 .with_min_is_exact(false)
@@ -848,45 +922,123 @@ mod tests {
             assert_eq!(from_thrift(tpe, thrift_stats).unwrap(), Some(stats));
         }
 
-        check_stats(Statistics::boolean(Some(false), Some(true), None, 7, true));
-        check_stats(Statistics::boolean(Some(false), Some(true), None, 7, true));
-        check_stats(Statistics::boolean(Some(false), Some(true), None, 0, false));
-        check_stats(Statistics::boolean(Some(true), Some(true), None, 7, true));
-        check_stats(Statistics::boolean(Some(false), Some(false), None, 7, true));
-        check_stats(Statistics::boolean(None, None, None, 7, true));
+        check_stats(Statistics::boolean(
+            Some(false),
+            Some(true),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::boolean(
+            Some(false),
+            Some(true),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::boolean(
+            Some(false),
+            Some(true),
+            None,
+            Some(0),
+            false,
+        ));
+        check_stats(Statistics::boolean(
+            Some(true),
+            Some(true),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::boolean(
+            Some(false),
+            Some(false),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::boolean(None, None, None, Some(7), true));
 
-        check_stats(Statistics::int32(Some(-100), Some(500), None, 7, true));
-        check_stats(Statistics::int32(Some(-100), Some(500), None, 0, false));
-        check_stats(Statistics::int32(None, None, None, 7, true));
+        check_stats(Statistics::int32(
+            Some(-100),
+            Some(500),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::int32(
+            Some(-100),
+            Some(500),
+            None,
+            Some(0),
+            false,
+        ));
+        check_stats(Statistics::int32(None, None, None, Some(7), true));
 
-        check_stats(Statistics::int64(Some(-100), Some(200), None, 7, true));
-        check_stats(Statistics::int64(Some(-100), Some(200), None, 0, false));
-        check_stats(Statistics::int64(None, None, None, 7, true));
+        check_stats(Statistics::int64(
+            Some(-100),
+            Some(200),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::int64(
+            Some(-100),
+            Some(200),
+            None,
+            Some(0),
+            false,
+        ));
+        check_stats(Statistics::int64(None, None, None, Some(7), true));
 
-        check_stats(Statistics::float(Some(1.2), Some(3.4), None, 7, true));
-        check_stats(Statistics::float(Some(1.2), Some(3.4), None, 0, false));
-        check_stats(Statistics::float(None, None, None, 7, true));
+        check_stats(Statistics::float(Some(1.2), Some(3.4), None, Some(7), true));
+        check_stats(Statistics::float(
+            Some(1.2),
+            Some(3.4),
+            None,
+            Some(0),
+            false,
+        ));
+        check_stats(Statistics::float(None, None, None, Some(7), true));
 
-        check_stats(Statistics::double(Some(1.2), Some(3.4), None, 7, true));
-        check_stats(Statistics::double(Some(1.2), Some(3.4), None, 0, false));
-        check_stats(Statistics::double(None, None, None, 7, true));
+        check_stats(Statistics::double(
+            Some(1.2),
+            Some(3.4),
+            None,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::double(
+            Some(1.2),
+            Some(3.4),
+            None,
+            Some(0),
+            false,
+        ));
+        check_stats(Statistics::double(None, None, None, Some(7), true));
 
         check_stats(Statistics::byte_array(
             Some(ByteArray::from(vec![1, 2, 3])),
             Some(ByteArray::from(vec![3, 4, 5])),
             None,
-            7,
+            Some(7),
             true,
         ));
-        check_stats(Statistics::byte_array(None, None, None, 7, true));
+        check_stats(Statistics::byte_array(None, None, None, Some(7), true));
 
         check_stats(Statistics::fixed_len_byte_array(
             Some(ByteArray::from(vec![1, 2, 3]).into()),
             Some(ByteArray::from(vec![3, 4, 5]).into()),
             None,
-            7,
+            Some(7),
+            true,
+        ));
+        check_stats(Statistics::fixed_len_byte_array(
+            None,
+            None,
+            None,
+            Some(7),
             true,
         ));
-        check_stats(Statistics::fixed_len_byte_array(None, None, None, 7, true));
     }
 }
diff --git a/parquet/src/file/writer.rs b/parquet/src/file/writer.rs
index c84d06a2c..a23a2a332 100644
--- a/parquet/src/file/writer.rs
+++ b/parquet/src/file/writer.rs
@@ -19,9 +19,8 @@
 //! using row group writers and column writers respectively.
 
 use crate::bloom_filter::Sbbf;
-use crate::file::page_index::index::Index;
 use crate::format as parquet;
-use crate::format::{ColumnIndex, OffsetIndex, RowGroup};
+use crate::format::{ColumnIndex, OffsetIndex};
 use crate::thrift::TSerializable;
 use std::fmt::Debug;
 use std::io::{BufWriter, IoSlice, Read};
@@ -38,7 +37,7 @@ use crate::errors::{ParquetError, Result};
 use crate::file::properties::{BloomFilterPosition, WriterPropertiesPtr};
 use crate::file::reader::ChunkReader;
 use crate::file::{metadata::*, PARQUET_MAGIC};
-use crate::schema::types::{self, ColumnDescPtr, SchemaDescPtr, SchemaDescriptor, TypePtr};
+use crate::schema::types::{ColumnDescPtr, SchemaDescPtr, SchemaDescriptor, TypePtr};
 
 /// A wrapper around a [`Write`] that keeps track of the number
 /// of bytes that have been written. The given [`Write`] is wrapped
@@ -730,281 +729,6 @@ impl<'a, W: Write + Send> PageWriter for SerializedPageWriter<'a, W> {
     }
 }
 
-/// Writes `crate::file::metadata` structures to a thrift encdoded byte streams
-///
-/// This structure handles the details of writing the various parts of parquet
-/// metadata into a byte stream. It is used to write the metadata into a
-/// parquet file and can also write metadata into other locations (such as a
-/// store of bytes).
-///
-/// This is somewhat trickey because the metadata is not store as a single inline
-/// thrift struture. It can have several "out of band" structures such as the OffsetIndex
-/// and BloomFilters which are stored separately whose locations are stored as offsets
-struct ThriftMetadataWriter<'a, W: Write> {
-    buf: &'a mut TrackedWrite<W>,
-    schema: &'a TypePtr,
-    schema_descr: &'a SchemaDescPtr,
-    row_groups: Vec<RowGroup>,
-    column_indexes: Option<&'a [Vec<Option<ColumnIndex>>]>,
-    offset_indexes: Option<&'a [Vec<Option<OffsetIndex>>]>,
-    key_value_metadata: Option<Vec<KeyValue>>,
-    created_by: Option<String>,
-    writer_version: i32,
-}
-
-impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
-    /// Serialize all the offset index to the file
-    fn write_offset_indexes(&mut self, offset_indexes: &[Vec<Option<OffsetIndex>>]) -> Result<()> {
-        // iter row group
-        // iter each column
-        // 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 => {}
-                }
-            }
-        }
-        Ok(())
-    }
-
-    /// Serialize all the column index to the file
-    fn write_column_indexes(&mut self, column_indexes: &[Vec<Option<ColumnIndex>>]) -> Result<()> {
-        // iter row group
-        // iter each column
-        // 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 => {}
-                }
-            }
-        }
-        Ok(())
-    }
-
-    /// Assembles and writes the final metadata to self.buf
-    pub fn finish(mut self) -> Result<parquet::FileMetaData> {
-        let num_rows = self.row_groups.iter().map(|x| x.num_rows).sum();
-
-        // Write column indexes and offset indexes
-        if let Some(column_indexes) = self.column_indexes {
-            self.write_column_indexes(column_indexes)?;
-        }
-        if let Some(offset_indexes) = self.offset_indexes {
-            self.write_offset_indexes(offset_indexes)?;
-        }
-
-        // We only include ColumnOrder for leaf nodes.
-        // Currently only supported ColumnOrder is TypeDefinedOrder so we set this
-        // for all leaf nodes.
-        // Even if the column has an undefined sort order, such as INTERVAL, this
-        // is still technically the defined TYPEORDER so it should still be set.
-        let column_orders = (0..self.schema_descr.num_columns())
-            .map(|_| parquet::ColumnOrder::TYPEORDER(parquet::TypeDefinedOrder {}))
-            .collect();
-        // This field is optional, perhaps in cases where no min/max fields are set
-        // in any Statistics or ColumnIndex object in the whole file.
-        // But for simplicity we always set this field.
-        let column_orders = Some(column_orders);
-
-        let file_metadata = parquet::FileMetaData {
-            num_rows,
-            row_groups: self.row_groups,
-            key_value_metadata: self.key_value_metadata.clone(),
-            version: self.writer_version,
-            schema: types::to_thrift(self.schema.as_ref())?,
-            created_by: self.created_by.clone(),
-            column_orders,
-            encryption_algorithm: None,
-            footer_signing_key_metadata: None,
-        };
-
-        // Write file metadata
-        let start_pos = self.buf.bytes_written();
-        {
-            let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
-            file_metadata.write_to_out_protocol(&mut protocol)?;
-        }
-        let end_pos = self.buf.bytes_written();
-
-        // Write footer
-        let metadata_len = (end_pos - start_pos) as u32;
-
-        self.buf.write_all(&metadata_len.to_le_bytes())?;
-        self.buf.write_all(&PARQUET_MAGIC)?;
-        Ok(file_metadata)
-    }
-
-    pub(self) fn new(
-        buf: &'a mut TrackedWrite<W>,
-        schema: &'a TypePtr,
-        schema_descr: &'a SchemaDescPtr,
-        row_groups: Vec<RowGroup>,
-        created_by: Option<String>,
-        writer_version: i32,
-    ) -> Self {
-        Self {
-            buf,
-            schema,
-            schema_descr,
-            row_groups,
-            column_indexes: None,
-            offset_indexes: None,
-            key_value_metadata: None,
-            created_by,
-            writer_version,
-        }
-    }
-
-    pub fn with_column_indexes(mut self, column_indexes: &'a [Vec<Option<ColumnIndex>>]) -> Self {
-        self.column_indexes = Some(column_indexes);
-        self
-    }
-
-    pub fn with_offset_indexes(mut self, offset_indexes: &'a [Vec<Option<OffsetIndex>>]) -> Self {
-        self.offset_indexes = Some(offset_indexes);
-        self
-    }
-
-    pub fn with_key_value_metadata(mut self, key_value_metadata: Vec<KeyValue>) -> Self {
-        self.key_value_metadata = Some(key_value_metadata);
-        self
-    }
-}
-
-pub struct ParquetMetadataWriter<'a, W: Write> {
-    buf: TrackedWrite<W>,
-    write_page_index: bool,
-    metadata: &'a ParquetMetaData,
-}
-
-impl<'a, W: Write> ParquetMetadataWriter<'a, W> {
-    pub fn new(buf: W, metadata: &'a ParquetMetaData) -> Self {
-        Self {
-            buf: TrackedWrite::new(buf),
-            write_page_index: true,
-            metadata,
-        }
-    }
-
-    pub fn write_page_index(&mut self, write_page_index: bool) -> &mut Self {
-        self.write_page_index = write_page_index;
-        self
-    }
-
-    pub fn finish(&mut self) -> Result<()> {
-        let file_metadata = self.metadata.file_metadata();
-
-        let schema = Arc::new(file_metadata.schema().clone());
-        let schema_descr = Arc::new(SchemaDescriptor::new(schema.clone()));
-        let created_by = file_metadata.created_by().map(str::to_string);
-
-        let row_groups = self
-            .metadata
-            .row_groups()
-            .iter()
-            .map(|rg| rg.to_thrift())
-            .collect::<Vec<_>>();
-
-        let key_value_metadata = file_metadata.key_value_metadata().cloned();
-
-        let column_indexes = self.convert_column_indexes();
-        let offset_indexes = self.convert_offset_index();
-
-        let mut encoder = ThriftMetadataWriter::new(
-            &mut self.buf,
-            &schema,
-            &schema_descr,
-            row_groups,
-            created_by,
-            file_metadata.version(),
-        );
-        encoder = encoder.with_column_indexes(&column_indexes);
-        encoder = encoder.with_offset_indexes(&offset_indexes);
-        if let Some(key_value_metadata) = key_value_metadata {
-            encoder = encoder.with_key_value_metadata(key_value_metadata);
-        }
-        encoder.finish()?;
-
-        Ok(())
-    }
-
-    fn convert_column_indexes(&self) -> Vec<Vec<Option<ColumnIndex>>> {
-        if let Some(row_group_column_indexes) = self.metadata.column_index() {
-            (0..self.metadata.row_groups().len())
-                .map(|rg_idx| {
-                    let column_indexes = &row_group_column_indexes[rg_idx];
-                    column_indexes
-                        .iter()
-                        .map(|column_index| match column_index {
-                            Index::NONE => None,
-                            Index::BOOLEAN(column_index) => Some(column_index.to_thrift()),
-                            Index::BYTE_ARRAY(column_index) => Some(column_index.to_thrift()),
-                            Index::DOUBLE(column_index) => Some(column_index.to_thrift()),
-                            Index::FIXED_LEN_BYTE_ARRAY(column_index) => {
-                                Some(column_index.to_thrift())
-                            }
-                            Index::FLOAT(column_index) => Some(column_index.to_thrift()),
-                            Index::INT32(column_index) => Some(column_index.to_thrift()),
-                            Index::INT64(column_index) => Some(column_index.to_thrift()),
-                            Index::INT96(column_index) => Some(column_index.to_thrift()),
-                        })
-                        .collect()
-                })
-                .collect()
-        } else {
-            // make a None for each row group, for each column
-            self.metadata
-                .row_groups()
-                .iter()
-                .map(|rg| std::iter::repeat(None).take(rg.columns().len()).collect())
-                .collect()
-        }
-    }
-
-    fn convert_offset_index(&self) -> Vec<Vec<Option<OffsetIndex>>> {
-        if let Some(row_group_offset_indexes) = self.metadata.offset_index() {
-            (0..self.metadata.row_groups().len())
-                .map(|rg_idx| {
-                    let offset_indexes = &row_group_offset_indexes[rg_idx];
-                    offset_indexes
-                        .iter()
-                        .map(|offset_index| Some(offset_index.to_thrift()))
-                        .collect()
-                })
-                .collect()
-        } else {
-            // make a None for each row group, for each column
-            self.metadata
-                .row_groups()
-                .iter()
-                .map(|rg| std::iter::repeat(None).take(rg.columns().len()).collect())
-                .collect()
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1036,6 +760,7 @@ mod tests {
     use crate::format::SortingColumn;
     use crate::record::{Row, RowAccessor};
     use crate::schema::parser::parse_message_type;
+    use crate::schema::types;
     use crate::schema::types::{ColumnDescriptor, ColumnPath};
     use crate::util::test_common::rand_gen::RandGen;
 
@@ -1397,7 +1122,7 @@ mod tests {
                 encoding: Encoding::DELTA_BINARY_PACKED,
                 def_level_encoding: Encoding::RLE,
                 rep_level_encoding: Encoding::RLE,
-                statistics: Some(Statistics::int32(Some(1), Some(3), None, 7, true)),
+                statistics: Some(Statistics::int32(Some(1), Some(3), None, Some(7), true)),
             },
             Page::DataPageV2 {
                 buf: Bytes::from(vec![4; 128]),
@@ -1408,7 +1133,7 @@ mod tests {
                 def_levels_byte_len: 24,
                 rep_levels_byte_len: 32,
                 is_compressed: false,
-                statistics: Some(Statistics::int32(Some(1), Some(3), None, 7, true)),
+                statistics: Some(Statistics::int32(Some(1), Some(3), None, Some(7), true)),
             },
         ];
 
@@ -1431,7 +1156,7 @@ mod tests {
                 encoding: Encoding::DELTA_BINARY_PACKED,
                 def_level_encoding: Encoding::RLE,
                 rep_level_encoding: Encoding::RLE,
-                statistics: Some(Statistics::int32(Some(1), Some(3), None, 7, true)),
+                statistics: Some(Statistics::int32(Some(1), Some(3), None, Some(7), true)),
             },
             Page::DataPageV2 {
                 buf: Bytes::from(vec![4; 128]),
@@ -2265,239 +1990,6 @@ mod tests {
         assert!(offset_index[0][0].unencoded_byte_array_data_bytes.is_none());
     }
 
-    #[cfg(feature = "async")]
-    mod async_tests {
-        use std::sync::Arc;
-
-        use crate::file::footer::parse_metadata;
-        use crate::file::properties::{EnabledStatistics, WriterProperties};
-        use crate::file::reader::{FileReader, SerializedFileReader};
-        use crate::file::writer::ParquetMetadataWriter;
-        use crate::{
-            arrow::ArrowWriter,
-            file::{page_index::index::Index, serialized_reader::ReadOptionsBuilder},
-        };
-        use arrow_array::{ArrayRef, Int32Array, RecordBatch};
-        use arrow_schema::{DataType as ArrowDataType, Field, Schema};
-        use bytes::{BufMut, Bytes, BytesMut};
-
-        use super::{ColumnChunkMetaData, ParquetMetaData, RowGroupMetaData};
-
-        struct TestMetadata {
-            #[allow(dead_code)]
-            file_size: usize,
-            metadata: ParquetMetaData,
-        }
-
-        fn has_page_index(metadata: &ParquetMetaData) -> bool {
-            match metadata.column_index() {
-                Some(column_index) => column_index
-                    .iter()
-                    .any(|rg_idx| rg_idx.iter().all(|col_idx| !matches!(col_idx, Index::NONE))),
-                None => false,
-            }
-        }
-
-        #[test]
-        fn test_roundtrip_parquet_metadata_without_page_index() {
-            // We currently don't have an ad-hoc ParquetMetadata loader that can load page indexes so
-            // we at least test round trip without them
-            let metadata = get_test_metadata(false, false);
-            assert!(!has_page_index(&metadata.metadata));
-
-            let mut buf = BytesMut::new().writer();
-            {
-                let mut writer = ParquetMetadataWriter::new(&mut buf, &metadata.metadata);
-                writer.finish().unwrap();
-            }
-
-            let data = buf.into_inner().freeze();
-
-            let decoded_metadata = parse_metadata(&data).unwrap();
-            assert!(!has_page_index(&metadata.metadata));
-
-            assert_eq!(metadata.metadata, decoded_metadata);
-        }
-
-        fn get_test_metadata(write_page_index: bool, read_page_index: bool) -> TestMetadata {
-            let mut buf = BytesMut::new().writer();
-            let schema: Arc<Schema> = Arc::new(Schema::new(vec![Field::new(
-                "a",
-                ArrowDataType::Int32,
-                true,
-            )]));
-
-            let a: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None, Some(2)]));
-
-            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 mut writer = ArrowWriter::try_new(&mut buf, schema, Some(writer_props)).unwrap();
-            writer.write(&batch).unwrap();
-            writer.close().unwrap();
-
-            let data = buf.into_inner().freeze();
-
-            let reader_opts = match read_page_index {
-                true => ReadOptionsBuilder::new().with_page_index().build(),
-                false => ReadOptionsBuilder::new().build(),
-            };
-            let reader = SerializedFileReader::new_with_options(data.clone(), reader_opts).unwrap();
-            let metadata = reader.metadata().clone();
-            TestMetadata {
-                file_size: data.len(),
-                metadata,
-            }
-        }
-
-        /// Temporary function so we can test loading metadata with page indexes
-        /// while we haven't fully figured out how to load it cleanly
-        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;
-
-            /// Adapt a `Bytes` to a `MetadataFetch` implementation.
-            struct AsyncBytes {
-                data: Bytes,
-            }
-
-            impl AsyncBytes {
-                fn new(data: Bytes) -> Self {
-                    Self { data }
-                }
-            }
-
-            impl MetadataFetch for AsyncBytes {
-                fn fetch(&mut self, range: Range<usize>) -> BoxFuture<'_, ParquetResult<Bytes>> {
-                    async move { Ok(self.data.slice(range.start..range.end)) }.boxed()
-                }
-            }
-
-            /// A `MetadataFetch` implementation that reads from a subset of the full data
-            /// while accepting ranges that address the full data.
-            struct MaskedBytes {
-                inner: Box<dyn MetadataFetch + Send>,
-                inner_range: Range<usize>,
-            }
-
-            impl MaskedBytes {
-                fn new(inner: Box<dyn MetadataFetch + Send>, inner_range: Range<usize>) -> Self {
-                    Self { inner, inner_range }
-                }
-            }
-
-            impl MetadataFetch for &mut MaskedBytes {
-                fn fetch(&mut self, range: Range<usize>) -> BoxFuture<'_, ParquetResult<Bytes>> {
-                    let inner_range = self.inner_range.clone();
-                    println!("inner_range: {:?}", inner_range);
-                    println!("range: {:?}", range);
-                    assert!(inner_range.start <= range.start && inner_range.end >= range.end);
-                    let range =
-                        range.start - self.inner_range.start..range.end - self.inner_range.start;
-                    self.inner.fetch(range)
-                }
-            }
-
-            let metadata_length = data.len();
-            let mut reader = MaskedBytes::new(
-                Box::new(AsyncBytes::new(data)),
-                file_size - metadata_length..file_size,
-            );
-            let metadata = MetadataLoader::load(&mut reader, file_size, None)
-                .await
-                .unwrap();
-            let loaded_metadata = metadata.finish();
-            let mut metadata = MetadataLoader::new(&mut reader, loaded_metadata);
-            metadata.load_page_index(true, true).await.unwrap();
-            metadata.finish()
-        }
-
-        fn check_columns_are_equivalent(left: &ColumnChunkMetaData, right: &ColumnChunkMetaData) {
-            assert_eq!(left.column_descr(), right.column_descr());
-            assert_eq!(left.encodings(), right.encodings());
-            assert_eq!(left.num_values(), right.num_values());
-            assert_eq!(left.compressed_size(), right.compressed_size());
-            assert_eq!(left.data_page_offset(), right.data_page_offset());
-            assert_eq!(left.statistics(), right.statistics());
-            assert_eq!(left.offset_index_length(), right.offset_index_length());
-            assert_eq!(left.column_index_length(), right.column_index_length());
-            assert_eq!(
-                left.unencoded_byte_array_data_bytes(),
-                right.unencoded_byte_array_data_bytes()
-            );
-        }
-
-        fn check_row_groups_are_equivalent(left: &RowGroupMetaData, right: &RowGroupMetaData) {
-            assert_eq!(left.num_rows(), right.num_rows());
-            assert_eq!(left.file_offset(), right.file_offset());
-            assert_eq!(left.total_byte_size(), right.total_byte_size());
-            assert_eq!(left.schema_descr(), right.schema_descr());
-            assert_eq!(left.num_columns(), right.num_columns());
-            left.columns()
-                .iter()
-                .zip(right.columns().iter())
-                .for_each(|(lc, rc)| {
-                    check_columns_are_equivalent(lc, rc);
-                });
-        }
-
-        #[tokio::test]
-        async fn test_encode_parquet_metadata_with_page_index() {
-            // Create a ParquetMetadata with page index information
-            let metadata = get_test_metadata(true, true);
-            assert!(has_page_index(&metadata.metadata));
-
-            let mut buf = BytesMut::new().writer();
-            {
-                let mut writer = ParquetMetadataWriter::new(&mut buf, &metadata.metadata);
-                writer.finish().unwrap();
-            }
-
-            let data = buf.into_inner().freeze();
-
-            let decoded_metadata = load_metadata_from_bytes(data.len(), data).await;
-
-            // Because the page index offsets will differ, compare invariant parts of the metadata
-            assert_eq!(
-                metadata.metadata.file_metadata(),
-                decoded_metadata.file_metadata()
-            );
-            assert_eq!(
-                metadata.metadata.column_index(),
-                decoded_metadata.column_index()
-            );
-            assert_eq!(
-                metadata.metadata.offset_index(),
-                decoded_metadata.offset_index()
-            );
-            assert_eq!(
-                metadata.metadata.num_row_groups(),
-                decoded_metadata.num_row_groups()
-            );
-
-            metadata
-                .metadata
-                .row_groups()
-                .iter()
-                .zip(decoded_metadata.row_groups().iter())
-                .for_each(|(left, right)| {
-                    check_row_groups_are_equivalent(left, right);
-                });
-        }
-    }
-
     #[test]
     #[cfg(feature = "arrow")]
     fn test_byte_stream_split_extended_roundtrip() {


Test patch:
diff --git a/arrow-flight/tests/common/fixture.rs b/arrow-flight/tests/common/fixture.rs
index 141879e2a..a666fa5d0 100644
--- a/arrow-flight/tests/common/fixture.rs
+++ b/arrow-flight/tests/common/fixture.rs
@@ -41,6 +41,7 @@ pub struct TestFixture {
 
 impl TestFixture {
     /// create a new test fixture from the server
+    #[allow(dead_code)]
     pub async fn new<T: FlightService>(test_server: FlightServiceServer<T>) -> Self {
         // let OS choose a free port
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
diff --git a/arrow-flight/tests/common/mod.rs b/arrow-flight/tests/common/mod.rs
index 85716e560..c4ac027c5 100644
--- a/arrow-flight/tests/common/mod.rs
+++ b/arrow-flight/tests/common/mod.rs
@@ -18,3 +18,4 @@
 pub mod fixture;
 pub mod server;
 pub mod trailers_layer;
+pub mod utils;
diff --git a/arrow-flight/tests/common/utils.rs b/arrow-flight/tests/common/utils.rs
new file mode 100644
index 000000000..0f70e4b31
--- /dev/null
+++ b/arrow-flight/tests/common/utils.rs
@@ -0,0 +1,118 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Common utilities for testing flight clients and servers
+
+use std::sync::Arc;
+
+use arrow_array::{
+    types::Int32Type, ArrayRef, BinaryViewArray, DictionaryArray, Float64Array, RecordBatch,
+    StringViewArray, UInt8Array,
+};
+use arrow_schema::{DataType, Field, Schema};
+
+/// Make a primitive batch for testing
+///
+/// Example:
+/// i: 0, 1, None, 3, 4
+/// f: 5.0, 4.0, None, 2.0, 1.0
+#[allow(dead_code)]
+pub fn make_primitive_batch(num_rows: usize) -> RecordBatch {
+    let i: UInt8Array = (0..num_rows)
+        .map(|i| {
+            if i == num_rows / 2 {
+                None
+            } else {
+                Some(i.try_into().unwrap())
+            }
+        })
+        .collect();
+
+    let f: Float64Array = (0..num_rows)
+        .map(|i| {
+            if i == num_rows / 2 {
+                None
+            } else {
+                Some((num_rows - i) as f64)
+            }
+        })
+        .collect();
+
+    RecordBatch::try_from_iter(vec![("i", Arc::new(i) as ArrayRef), ("f", Arc::new(f))]).unwrap()
+}
+
+/// Make a dictionary batch for testing
+///
+/// Example:
+/// a: value0, value1, value2, None, value1, value2
+#[allow(dead_code)]
+pub fn make_dictionary_batch(num_rows: usize) -> RecordBatch {
+    let values: Vec<_> = (0..num_rows)
+        .map(|i| {
+            if i == num_rows / 2 {
+                None
+            } else {
+                // repeat some values for low cardinality
+                let v = i / 3;
+                Some(format!("value{v}"))
+            }
+        })
+        .collect();
+
+    let a: DictionaryArray<Int32Type> = values
+        .iter()
+        .map(|s| s.as_ref().map(|s| s.as_str()))
+        .collect();
+
+    RecordBatch::try_from_iter(vec![("a", Arc::new(a) as ArrayRef)]).unwrap()
+}
+
+#[allow(dead_code)]
+pub fn make_view_batches(num_rows: usize) -> RecordBatch {
+    const LONG_TEST_STRING: &str =
+        "This is a long string to make sure binary view array handles it";
+    let schema = Schema::new(vec![
+        Field::new("field1", DataType::BinaryView, true),
+        Field::new("field2", DataType::Utf8View, true),
+    ]);
+
+    let string_view_values: Vec<Option<&str>> = (0..num_rows)
+        .map(|i| match i % 3 {
+            0 => None,
+            1 => Some("foo"),
+            2 => Some(LONG_TEST_STRING),
+            _ => unreachable!(),
+        })
+        .collect();
+
+    let bin_view_values: Vec<Option<&[u8]>> = (0..num_rows)
+        .map(|i| match i % 3 {
+            0 => None,
+            1 => Some("bar".as_bytes()),
+            2 => Some(LONG_TEST_STRING.as_bytes()),
+            _ => unreachable!(),
+        })
+        .collect();
+
+    let binary_array = BinaryViewArray::from_iter(bin_view_values);
+    let utf8_array = StringViewArray::from_iter(string_view_values);
+    RecordBatch::try_new(
+        Arc::new(schema.clone()),
+        vec![Arc::new(binary_array), Arc::new(utf8_array)],
+    )
+    .unwrap()
+}
diff --git a/arrow-flight/tests/encode_decode.rs b/arrow-flight/tests/encode_decode.rs
index 0185fa77f..cbfae1825 100644
--- a/arrow-flight/tests/encode_decode.rs
+++ b/arrow-flight/tests/encode_decode.rs
@@ -19,11 +19,7 @@
 
 use std::{collections::HashMap, sync::Arc};
 
-use arrow_array::types::Int32Type;
-use arrow_array::{
-    ArrayRef, BinaryViewArray, DictionaryArray, Float64Array, RecordBatch, StringViewArray,
-    UInt8Array,
-};
+use arrow_array::{ArrayRef, RecordBatch};
 use arrow_cast::pretty::pretty_format_batches;
 use arrow_flight::flight_descriptor::DescriptorType;
 use arrow_flight::FlightDescriptor;
@@ -36,6 +32,9 @@ use arrow_schema::{DataType, Field, Fields, Schema, SchemaRef};
 use bytes::Bytes;
 use futures::{StreamExt, TryStreamExt};
 
+mod common;
+use common::utils::{make_dictionary_batch, make_primitive_batch, make_view_batches};
+
 #[tokio::test]
 async fn test_empty() {
     roundtrip(vec![]).await;
@@ -415,95 +414,6 @@ async fn test_mismatched_schema_message() {
     .await;
 }
 
-/// Make a primitive batch for testing
-///
-/// Example:
-/// i: 0, 1, None, 3, 4
-/// f: 5.0, 4.0, None, 2.0, 1.0
-fn make_primitive_batch(num_rows: usize) -> RecordBatch {
-    let i: UInt8Array = (0..num_rows)
-        .map(|i| {
-            if i == num_rows / 2 {
-                None
-            } else {
-                Some(i.try_into().unwrap())
-            }
-        })
-        .collect();
-
-    let f: Float64Array = (0..num_rows)
-        .map(|i| {
-            if i == num_rows / 2 {
-                None
-            } else {
-                Some((num_rows - i) as f64)
-            }
-        })
-        .collect();
-
-    RecordBatch::try_from_iter(vec![("i", Arc::new(i) as ArrayRef), ("f", Arc::new(f))]).unwrap()
-}
-
-/// Make a dictionary batch for testing
-///
-/// Example:
-/// a: value0, value1, value2, None, value1, value2
-fn make_dictionary_batch(num_rows: usize) -> RecordBatch {
-    let values: Vec<_> = (0..num_rows)
-        .map(|i| {
-            if i == num_rows / 2 {
-                None
-            } else {
-                // repeat some values for low cardinality
-                let v = i / 3;
-                Some(format!("value{v}"))
-            }
-        })
-        .collect();
-
-    let a: DictionaryArray<Int32Type> = values
-        .iter()
-        .map(|s| s.as_ref().map(|s| s.as_str()))
-        .collect();
-
-    RecordBatch::try_from_iter(vec![("a", Arc::new(a) as ArrayRef)]).unwrap()
-}
-
-fn make_view_batches(num_rows: usize) -> RecordBatch {
-    const LONG_TEST_STRING: &str =
-        "This is a long string to make sure binary view array handles it";
-    let schema = Schema::new(vec![
-        Field::new("field1", DataType::BinaryView, true),
-        Field::new("field2", DataType::Utf8View, true),
-    ]);
-
-    let string_view_values: Vec<Option<&str>> = (0..num_rows)
-        .map(|i| match i % 3 {
-            0 => None,
-            1 => Some("foo"),
-            2 => Some(LONG_TEST_STRING),
-            _ => unreachable!(),
-        })
-        .collect();
-
-    let bin_view_values: Vec<Option<&[u8]>> = (0..num_rows)
-        .map(|i| match i % 3 {
-            0 => None,
-            1 => Some("bar".as_bytes()),
-            2 => Some(LONG_TEST_STRING.as_bytes()),
-            _ => unreachable!(),
-        })
-        .collect();
-
-    let binary_array = BinaryViewArray::from_iter(bin_view_values);
-    let utf8_array = StringViewArray::from_iter(string_view_values);
-    RecordBatch::try_new(
-        Arc::new(schema.clone()),
-        vec![Arc::new(binary_array), Arc::new(utf8_array)],
-    )
-    .unwrap()
-}
-
 /// Encodes input as a FlightData stream, and then decodes it using
 /// FlightRecordBatchStream and validates the decoded record batches
 /// match the input.
diff --git a/arrow-flight/tests/flight_sql_client.rs b/arrow-flight/tests/flight_sql_client.rs
index 94b768a13..349da062a 100644
--- a/arrow-flight/tests/flight_sql_client.rs
+++ b/arrow-flight/tests/flight_sql_client.rs
@@ -18,14 +18,21 @@
 mod common;
 
 use crate::common::fixture::TestFixture;
+use crate::common::utils::make_primitive_batch;
+
+use arrow_array::RecordBatch;
+use arrow_flight::decode::FlightRecordBatchStream;
+use arrow_flight::error::FlightError;
 use arrow_flight::flight_service_server::FlightServiceServer;
 use arrow_flight::sql::client::FlightSqlServiceClient;
-use arrow_flight::sql::server::FlightSqlService;
+use arrow_flight::sql::server::{FlightSqlService, PeekableFlightDataStream};
 use arrow_flight::sql::{
     ActionBeginTransactionRequest, ActionBeginTransactionResult, ActionEndTransactionRequest,
-    EndTransaction, SqlInfo,
+    CommandStatementIngest, EndTransaction, SqlInfo, TableDefinitionOptions, TableExistsOption,
+    TableNotExistOption,
 };
 use arrow_flight::Action;
+use futures::{StreamExt, TryStreamExt};
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::Mutex;
@@ -34,9 +41,7 @@ use uuid::Uuid;
 
 #[tokio::test]
 pub async fn test_begin_end_transaction() {
-    let test_server = FlightSqlServiceImpl {
-        transactions: Arc::new(Mutex::new(HashMap::new())),
-    };
+    let test_server = FlightSqlServiceImpl::new();
     let fixture = TestFixture::new(test_server.service()).await;
     let channel = fixture.channel().await;
     let mut flight_sql_client = FlightSqlServiceClient::new(channel);
@@ -63,12 +68,83 @@ pub async fn test_begin_end_transaction() {
         .is_err());
 }
 
+#[tokio::test]
+pub async fn test_execute_ingest() {
+    let test_server = FlightSqlServiceImpl::new();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let channel = fixture.channel().await;
+    let mut flight_sql_client = FlightSqlServiceClient::new(channel);
+    let cmd = make_ingest_command();
+    let expected_rows = 10;
+    let batches = vec![
+        make_primitive_batch(5),
+        make_primitive_batch(3),
+        make_primitive_batch(2),
+    ];
+    let actual_rows = flight_sql_client
+        .execute_ingest(cmd, futures::stream::iter(batches.clone()).map(Ok))
+        .await
+        .expect("ingest should succeed");
+    assert_eq!(actual_rows, expected_rows);
+    // make sure the batches made it through to the server
+    let ingested_batches = test_server.ingested_batches.lock().await.clone();
+    assert_eq!(ingested_batches, batches);
+}
+
+#[tokio::test]
+pub async fn test_execute_ingest_error() {
+    let test_server = FlightSqlServiceImpl::new();
+    let fixture = TestFixture::new(test_server.service()).await;
+    let channel = fixture.channel().await;
+    let mut flight_sql_client = FlightSqlServiceClient::new(channel);
+    let cmd = make_ingest_command();
+    // send an error from the client
+    let batches = vec![
+        Ok(make_primitive_batch(5)),
+        Err(FlightError::NotYetImplemented(
+            "Client error message".to_string(),
+        )),
+    ];
+    // make sure the client returns the error from the client
+    let err = flight_sql_client
+        .execute_ingest(cmd, futures::stream::iter(batches))
+        .await
+        .unwrap_err();
+    assert_eq!(
+        err.to_string(),
+        "External error: Not yet implemented: Client error message"
+    );
+}
+
+fn make_ingest_command() -> CommandStatementIngest {
+    CommandStatementIngest {
+        table_definition_options: Some(TableDefinitionOptions {
+            if_not_exist: TableNotExistOption::Create.into(),
+            if_exists: TableExistsOption::Fail.into(),
+        }),
+        table: String::from("test"),
+        schema: None,
+        catalog: None,
+        temporary: true,
+        transaction_id: None,
+        options: HashMap::default(),
+    }
+}
+
 #[derive(Clone)]
 pub struct FlightSqlServiceImpl {
     transactions: Arc<Mutex<HashMap<String, ()>>>,
+    ingested_batches: Arc<Mutex<Vec<RecordBatch>>>,
 }
 
 impl FlightSqlServiceImpl {
+    pub fn new() -> Self {
+        Self {
+            transactions: Arc::new(Mutex::new(HashMap::new())),
+            ingested_batches: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+
     /// Return an [`FlightServiceServer`] that can be used with a
     /// [`Server`](tonic::transport::Server)
     pub fn service(&self) -> FlightServiceServer<Self> {
@@ -77,6 +153,12 @@ impl FlightSqlServiceImpl {
     }
 }
 
+impl Default for FlightSqlServiceImpl {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[tonic::async_trait]
 impl FlightSqlService for FlightSqlServiceImpl {
     type FlightService = FlightSqlServiceImpl;
@@ -116,4 +198,19 @@ impl FlightSqlService for FlightSqlServiceImpl {
     }
 
     async fn register_sql_info(&self, _id: i32, _result: &SqlInfo) {}
+
+    async fn do_put_statement_ingest(
+        &self,
+        _ticket: CommandStatementIngest,
+        request: Request<PeekableFlightDataStream>,
+    ) -> Result<i64, Status> {
+        let batches: Vec<RecordBatch> = FlightRecordBatchStream::new_from_flight_data(
+            request.into_inner().map_err(|e| e.into()),
+        )
+        .try_collect()
+        .await?;
+        let affected_rows = batches.iter().map(|batch| batch.num_rows() as i64).sum();
+        *self.ingested_batches.lock().await.as_mut() = batches;
+        Ok(affected_rows)
+    }
 }