Repository: apache/arrow-rs
Pull request #6232 (2024-08-13 04:56:31)

Avoid infinite loop in bad parquet by checking the number of rep levels 
# Which issue does this PR close?

This fixes one case of #6228.

# Rationale for this change
 
In read_records, the number of rep levels read from page can be 0 if the file is corrupted. But currently we don't check that and in that case the execution will stuck in the read_records, keeping reading but reading nothing.

# What changes are included in this PR?

Error out if reading 0 rep level.

# Are there any user-facing changes?

not really.



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/reader.rs b/parquet/src/column/reader.rs
index b40ca2b78..0c7cbb412 100644
--- a/parquet/src/column/reader.rs
+++ b/parquet/src/column/reader.rs
@@ -240,6 +240,12 @@ where
                     let (mut records_read, levels_read) =
                         reader.read_rep_levels(out, remaining_records, remaining_levels)?;
 
+                    if records_read == 0 && levels_read == 0 {
+                        // The fact that we're still looping implies there must be some levels to read.
+                        return Err(general_err!(
+                            "Insufficient repetition levels read from column"
+                        ));
+                    }
                     if levels_read == remaining_levels && self.has_record_delimiter {
                         // Reached end of page, which implies records_read < remaining_records
                         // as otherwise would have stopped reading before reaching the end
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 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)
+    }
 }