Repository: apache/arrow-rs
Pull request #5972 (2024-06-27 20:41:02)

Implement directly build byte view array on top of parquet buffer
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
~This PR is not ready to review until we merged #5970  .~

Part of https://github.com/apache/arrow-rs/issues/5904 , sequel to https://github.com/apache/arrow-rs/pull/5968 and #5970


# Rationale for this change
 This PR has the real work to directly transform the parquet page buffer into a view buffer without extra copy.
To see the performance difference, you can run:
```bash
cargo bench --bench arrow_reader --features="arrow test_common experimental" "arrow_array_reader/Binary.*Array/plain encoded"
```

```
arrow_array_reader/BinaryArray/plain encoded, mandatory, no NULLs
                        time:   [321.33 µs 322.06 µs 323.48 µs]
                        change: [-1.1124% -0.8542% -0.4913%] (p = 0.00 < 0.05)
                        Change within noise threshold.
arrow_array_reader/BinaryArray/plain encoded, optional, no NULLs
                        time:   [327.34 µs 327.65 µs 328.06 µs]
                        change: [-0.8087% -0.7037% -0.5861%] (p = 0.00 < 0.05)
                        Change within noise threshold.
arrow_array_reader/BinaryArray/plain encoded, optional, half NULLs
                        time:   [416.00 µs 416.32 µs 416.65 µs]
                        change: [+0.4281% +0.6535% +0.8394%] (p = 0.00 < 0.05)
                        Change within noise threshold.

arrow_array_reader/BinaryViewArray/plain encoded, mandatory, no NULLs
                        time:   [238.51 µs 239.00 µs 239.40 µs]
                        change: [-3.9669% -3.6590% -3.3511%] (p = 0.00 < 0.05)
                        Performance has improved.
arrow_array_reader/BinaryViewArray/plain encoded, optional, no NULLs
                        time:   [243.73 µs 243.82 µs 243.91 µs]
                        change: [+1.3771% +1.5136% +1.6413%] (p = 0.00 < 0.05)
                        Performance has regressed.
arrow_array_reader/BinaryViewArray/plain encoded, optional, half NULLs
                        time:   [179.69 µs 180.14 µs 180.87 µs]
                        change: [-0.7692% -0.4983% -0.1523%] (p = 0.00 < 0.05)
                        Change within noise threshold.
```

You should find that with this PR, reading BinaryViewArray is faster than reading BinaryArray -- a milestone from making StringViewArray faster than StringArray.

When this (set of) PR is merged, the last piece is to make utf8 validation fast, so that string view can maintain the advantage.
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

# What changes are included in this PR?
This PR only includes decoding plain data for ease of review. Supporting for RLE/dictionary will be filed soon.

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml
index accb9f881..08d287bcc 100644
--- a/.github/workflows/docs.yml
+++ b/.github/workflows/docs.yml
@@ -95,3 +95,5 @@ jobs:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           publish_dir: website/build
           publish_branch: asf-site
+          # Avoid accumulating history of in progress API jobs: https://github.com/apache/arrow-rs/issues/5908
+          force_orphan: true
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index c18509aaf..5a0df233c 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -46,7 +46,7 @@ walkdir = "2"
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
-quick-xml = { version = "0.34.0", features = ["serialize", "overlapped-lists"], optional = true }
+quick-xml = { version = "0.35.0", features = ["serialize", "overlapped-lists"], optional = true }
 serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
 serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
diff --git a/object_store/src/attributes.rs b/object_store/src/attributes.rs
index ecef32eb4..11cf27c85 100644
--- a/object_store/src/attributes.rs
+++ b/object_store/src/attributes.rs
@@ -45,6 +45,10 @@ pub enum Attribute {
     ///
     /// See [Cache-Control](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Cache-Control)
     CacheControl,
+    /// Specifies a user-defined metadata field for the object
+    ///
+    /// The String is a user-defined key
+    Metadata(Cow<'static, str>),
 }
 
 /// The value of an [`Attribute`]
@@ -194,10 +198,11 @@ mod tests {
             (Attribute::ContentLanguage, "en-US"),
             (Attribute::ContentType, "test"),
             (Attribute::CacheControl, "control"),
+            (Attribute::Metadata("key1".into()), "value1"),
         ]);
 
         assert!(!attributes.is_empty());
-        assert_eq!(attributes.len(), 5);
+        assert_eq!(attributes.len(), 6);
 
         assert_eq!(
             attributes.get(&Attribute::ContentType),
@@ -210,18 +215,18 @@ mod tests {
             attributes.insert(Attribute::CacheControl, "v1".into()),
             Some(metav)
         );
-        assert_eq!(attributes.len(), 5);
+        assert_eq!(attributes.len(), 6);
 
         assert_eq!(
             attributes.remove(&Attribute::CacheControl).unwrap(),
             "v1".into()
         );
-        assert_eq!(attributes.len(), 4);
+        assert_eq!(attributes.len(), 5);
 
         let metav: AttributeValue = "v2".into();
         attributes.insert(Attribute::CacheControl, metav.clone());
         assert_eq!(attributes.get(&Attribute::CacheControl), Some(&metav));
-        assert_eq!(attributes.len(), 5);
+        assert_eq!(attributes.len(), 6);
 
         assert_eq!(
             attributes.get(&Attribute::ContentDisposition),
@@ -235,5 +240,9 @@ mod tests {
             attributes.get(&Attribute::ContentLanguage),
             Some(&"en-US".into())
         );
+        assert_eq!(
+            attributes.get(&Attribute::Metadata("key1".into())),
+            Some(&"value1".into())
+        );
     }
 }
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 98226c4b6..ab4da86f5 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -61,6 +61,7 @@ use std::sync::Arc;
 
 const VERSION_HEADER: &str = "x-amz-version-id";
 const SHA256_CHECKSUM: &str = "x-amz-checksum-sha256";
+const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-amz-meta-";
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
@@ -326,6 +327,10 @@ impl<'a> Request<'a> {
                     has_content_type = true;
                     builder.header(CONTENT_TYPE, v.as_ref())
                 }
+                Attribute::Metadata(k_suffix) => builder.header(
+                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
+                    v.as_ref(),
+                ),
             };
         }
 
@@ -642,6 +647,7 @@ impl GetClient for S3Client {
         etag_required: false,
         last_modified_required: false,
         version_header: Some(VERSION_HEADER),
+        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
     };
 
     /// Make an S3 GET request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html>
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index be760c7dd..b5e82c2a8 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -48,6 +48,7 @@ use std::time::Duration;
 use url::Url;
 
 const VERSION_HEADER: &str = "x-ms-version-id";
+const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-ms-meta-";
 static MS_CACHE_CONTROL: HeaderName = HeaderName::from_static("x-ms-blob-cache-control");
 static MS_CONTENT_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-content-type");
 static MS_CONTENT_DISPOSITION: HeaderName =
@@ -208,6 +209,10 @@ impl<'a> PutRequest<'a> {
                     has_content_type = true;
                     builder.header(&MS_CONTENT_TYPE, v.as_ref())
                 }
+                Attribute::Metadata(k_suffix) => builder.header(
+                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
+                    v.as_ref(),
+                ),
             };
         }
 
@@ -499,6 +504,7 @@ impl GetClient for AzureClient {
         etag_required: true,
         last_modified_required: true,
         version_header: Some(VERSION_HEADER),
+        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
     };
 
     /// Make an Azure GET request
diff --git a/object_store/src/buffered.rs b/object_store/src/buffered.rs
index c7b71aa1c..fcd7e064e 100644
--- a/object_store/src/buffered.rs
+++ b/object_store/src/buffered.rs
@@ -438,7 +438,12 @@ impl AsyncWrite for BufWriter {
                 }
                 BufWriterState::Flush(f) => return f.poll_unpin(cx).map_err(std::io::Error::from),
                 BufWriterState::Write(x) => {
-                    let upload = x.take().unwrap();
+                    let upload = x.take().ok_or_else(|| {
+                        std::io::Error::new(
+                            ErrorKind::InvalidInput,
+                            "Cannot shutdown a writer that has already been shut down",
+                        )
+                    })?;
                     self.state = BufWriterState::Flush(
                         async move {
                             upload.finish().await?;
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 430b87b1b..b45eaa143 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -135,6 +135,9 @@ enum GetResultError {
     #[snafu(display("Content-Type header contained non UTF-8 characters"))]
     InvalidContentType { source: ToStrError },
 
+    #[snafu(display("Metadata value for \"{key:?}\" contained non UTF-8 characters"))]
+    InvalidMetadata { key: String },
+
     #[snafu(display("Requested {expected:?}, got {actual:?}"))]
     UnexpectedRange {
         expected: Range<usize>,
@@ -192,7 +195,7 @@ fn get_result<T: GetClient>(
         }}
     }
 
-    let attributes = parse_attributes!(
+    let mut attributes = parse_attributes!(
         response.headers(),
         (
             CACHE_CONTROL,
@@ -221,6 +224,24 @@ fn get_result<T: GetClient>(
         )
     );
 
+    // Add attributes that match the user-defined metadata prefix (e.g. x-amz-meta-)
+    if let Some(prefix) = T::HEADER_CONFIG.user_defined_metadata_prefix {
+        for (key, val) in response.headers() {
+            if let Some(suffix) = key.as_str().strip_prefix(prefix) {
+                if let Ok(val_str) = val.to_str() {
+                    attributes.insert(
+                        Attribute::Metadata(suffix.to_string().into()),
+                        val_str.to_string().into(),
+                    );
+                } else {
+                    return Err(GetResultError::InvalidMetadata {
+                        key: key.to_string(),
+                    });
+                }
+            }
+        }
+    }
+
     let stream = response
         .bytes_stream()
         .map_err(|source| crate::Error::Generic {
@@ -253,6 +274,7 @@ mod tests {
             etag_required: false,
             last_modified_required: false,
             version_header: None,
+            user_defined_metadata_prefix: Some("x-test-meta-"),
         };
 
         async fn get_request(&self, _: &Path, _: GetOptions) -> Result<Response> {
@@ -265,6 +287,7 @@ mod tests {
         range: Option<Range<usize>>,
         status: StatusCode,
         content_range: Option<&str>,
+        headers: Option<Vec<(&str, &str)>>,
     ) -> Response {
         let mut builder = http::Response::builder();
         if let Some(range) = content_range {
@@ -276,6 +299,12 @@ mod tests {
             None => vec![0_u8; object_size],
         };
 
+        if let Some(headers) = headers {
+            for (key, value) in headers {
+                builder = builder.header(key, value);
+            }
+        }
+
         builder
             .status(status)
             .header(CONTENT_LENGTH, object_size)
@@ -288,7 +317,7 @@ mod tests {
     async fn test_get_result() {
         let path = Path::from("test");
 
-        let resp = make_response(12, None, StatusCode::OK, None);
+        let resp = make_response(12, None, StatusCode::OK, None, None);
         let res = get_result::<TestClient>(&path, None, resp).unwrap();
         assert_eq!(res.meta.size, 12);
         assert_eq!(res.range, 0..12);
@@ -302,6 +331,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-2/12"),
+            None,
         );
         let res = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap();
         assert_eq!(res.meta.size, 12);
@@ -309,7 +339,7 @@ mod tests {
         let bytes = res.bytes().await.unwrap();
         assert_eq!(bytes.len(), 1);
 
-        let resp = make_response(12, Some(2..3), StatusCode::OK, None);
+        let resp = make_response(12, Some(2..3), StatusCode::OK, None, None);
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
             err.to_string(),
@@ -321,6 +351,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-3/12"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(err.to_string(), "Requested 2..3, got 2..4");
@@ -330,6 +361,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-2/*"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
@@ -337,7 +369,7 @@ mod tests {
             "Failed to parse value for CONTENT_RANGE header: \"bytes 2-2/*\""
         );
 
-        let resp = make_response(12, Some(2..3), StatusCode::PARTIAL_CONTENT, None);
+        let resp = make_response(12, Some(2..3), StatusCode::PARTIAL_CONTENT, None, None);
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
             err.to_string(),
@@ -349,6 +381,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-3/2"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
@@ -361,6 +394,7 @@ mod tests {
             Some(2..6),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-5/6"),
+            None,
         );
         let res = get_result::<TestClient>(&path, Some(GetRange::Suffix(4)), resp).unwrap();
         assert_eq!(res.meta.size, 6);
@@ -373,8 +407,26 @@ mod tests {
             Some(2..6),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-3/6"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(GetRange::Suffix(4)), resp).unwrap_err();
         assert_eq!(err.to_string(), "Requested 2..6, got 2..4");
+
+        let resp = make_response(
+            12,
+            None,
+            StatusCode::OK,
+            None,
+            Some(vec![("x-test-meta-foo", "bar")]),
+        );
+        let res = get_result::<TestClient>(&path, None, resp).unwrap();
+        assert_eq!(res.meta.size, 12);
+        assert_eq!(res.range, 0..12);
+        assert_eq!(
+            res.attributes.get(&Attribute::Metadata("foo".into())),
+            Some(&"bar".into())
+        );
+        let bytes = res.bytes().await.unwrap();
+        assert_eq!(bytes.len(), 12);
     }
 }
diff --git a/object_store/src/client/header.rs b/object_store/src/client/header.rs
index e85bf6ba5..9ce5db433 100644
--- a/object_store/src/client/header.rs
+++ b/object_store/src/client/header.rs
@@ -31,6 +31,7 @@ pub struct HeaderConfig {
     ///
     /// Defaults to `true`
     pub etag_required: bool,
+
     /// Whether to require a Last-Modified header when extracting [`ObjectMeta`] from headers.
     ///
     /// Defaults to `true`
@@ -38,6 +39,9 @@ pub struct HeaderConfig {
 
     /// The version header name if any
     pub version_header: Option<&'static str>,
+
+    /// The user defined metadata prefix if any
+    pub user_defined_metadata_prefix: Option<&'static str>,
 }
 
 #[derive(Debug, Snafu)]
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index 35c64cc3d..0045383ee 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -49,6 +49,7 @@ use std::sync::Arc;
 
 const VERSION_HEADER: &str = "x-goog-generation";
 const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
+const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-goog-meta-";
 
 static VERSION_MATCH: HeaderName = HeaderName::from_static("x-goog-if-generation-match");
 
@@ -199,6 +200,10 @@ impl<'a> Request<'a> {
                     has_content_type = true;
                     builder.header(CONTENT_TYPE, v.as_ref())
                 }
+                Attribute::Metadata(k_suffix) => builder.header(
+                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
+                    v.as_ref(),
+                ),
             };
         }
 
@@ -567,6 +572,7 @@ impl GetClient for GoogleCloudStorageClient {
         etag_required: true,
         last_modified_required: true,
         version_header: Some(VERSION_HEADER),
+        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
     };
 
     /// Perform a get request <https://cloud.google.com/storage/docs/xml-api/get-object-download>
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index 4dccef880..5def93120 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -183,6 +183,8 @@ impl Client {
                         has_content_type = true;
                         builder.header(CONTENT_TYPE, v.as_ref())
                     }
+                    // Ignore metadata attributes
+                    Attribute::Metadata(_) => builder,
                 };
             }
 
@@ -319,6 +321,7 @@ impl GetClient for Client {
         etag_required: false,
         last_modified_required: false,
         version_header: None,
+        user_defined_metadata_prefix: None,
     };
 
     async fn get_request(&self, path: &Path, options: GetOptions) -> Result<Response> {
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
index 31b074f49..89b21bc61 100644
--- a/object_store/src/integration.rs
+++ b/object_store/src/integration.rs
@@ -480,6 +480,7 @@ pub async fn put_get_attributes(integration: &dyn ObjectStore) {
         (Attribute::ContentEncoding, "gzip"),
         (Attribute::ContentLanguage, "en-US"),
         (Attribute::ContentType, "text/html; charset=utf-8"),
+        (Attribute::Metadata("test_key".into()), "test_value"),
     ]);
 
     let path = Path::from("attributes");
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 1ce588af2..d3bfab8ed 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -724,7 +724,7 @@ struct LocalUpload {
 #[derive(Debug)]
 struct UploadState {
     dest: PathBuf,
-    file: Mutex<Option<File>>,
+    file: Mutex<File>,
 }
 
 impl LocalUpload {
@@ -732,7 +732,7 @@ impl LocalUpload {
         Self {
             state: Arc::new(UploadState {
                 dest,
-                file: Mutex::new(Some(file)),
+                file: Mutex::new(file),
             }),
             src: Some(src),
             offset: 0,
@@ -748,8 +748,7 @@ impl MultipartUpload for LocalUpload {
 
         let s = Arc::clone(&self.state);
         maybe_spawn_blocking(move || {
-            let mut f = s.file.lock();
-            let file = f.as_mut().context(AbortedSnafu)?;
+            let mut file = s.file.lock();
             file.seek(SeekFrom::Start(offset))
                 .context(SeekSnafu { path: &s.dest })?;
 
@@ -767,9 +766,9 @@ impl MultipartUpload for LocalUpload {
         let s = Arc::clone(&self.state);
         maybe_spawn_blocking(move || {
             // Ensure no inflight writes
-            let f = s.file.lock().take().context(AbortedSnafu)?;
+            let file = s.file.lock();
             std::fs::rename(&src, &s.dest).context(UnableToRenameFileSnafu)?;
-            let metadata = f.metadata().map_err(|e| Error::Metadata {
+            let metadata = file.metadata().map_err(|e| Error::Metadata {
                 source: e.into(),
                 path: src.to_string_lossy().to_string(),
             })?;
diff --git a/object_store/src/upload.rs b/object_store/src/upload.rs
index dc499e2e0..4df4d8fd4 100644
--- a/object_store/src/upload.rs
+++ b/object_store/src/upload.rs
@@ -226,7 +226,15 @@ impl WriteMultipart {
         }
 
         self.wait_for_capacity(0).await?;
-        self.upload.complete().await
+
+        match self.upload.complete().await {
+            Err(e) => {
+                self.tasks.shutdown().await;
+                self.upload.abort().await?;
+                Err(e)
+            }
+            Ok(result) => Ok(result),
+        }
     }
 }
 
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
new file mode 100644
index 000000000..2d4e57414
--- /dev/null
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -0,0 +1,427 @@
+// 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::arrow::array_reader::{read_records, skip_records, ArrayReader};
+use crate::arrow::buffer::view_buffer::ViewBuffer;
+use crate::arrow::record_reader::GenericRecordReader;
+use crate::arrow::schema::parquet_to_arrow_field;
+use crate::basic::{ConvertedType, Encoding};
+use crate::column::page::PageIterator;
+use crate::column::reader::decoder::ColumnValueDecoder;
+use crate::errors::{ParquetError, Result};
+use crate::schema::types::ColumnDescPtr;
+use arrow_array::ArrayRef;
+use arrow_schema::DataType as ArrowType;
+use bytes::Bytes;
+use std::any::Any;
+
+/// Returns an [`ArrayReader`] that decodes the provided byte array column to view types.
+#[allow(unused)]
+pub fn make_byte_view_array_reader(
+    pages: Box<dyn PageIterator>,
+    column_desc: ColumnDescPtr,
+    arrow_type: Option<ArrowType>,
+) -> Result<Box<dyn ArrayReader>> {
+    // Check if Arrow type is specified, else create it from Parquet type
+    let data_type = match arrow_type {
+        Some(t) => t,
+        None => match parquet_to_arrow_field(column_desc.as_ref())?.data_type() {
+            ArrowType::Utf8 | ArrowType::Utf8View => ArrowType::Utf8View,
+            _ => ArrowType::BinaryView,
+        },
+    };
+
+    match data_type {
+        ArrowType::BinaryView | ArrowType::Utf8View => {
+            let reader = GenericRecordReader::new(column_desc);
+            Ok(Box::new(ByteViewArrayReader::new(pages, data_type, reader)))
+        }
+
+        _ => Err(general_err!(
+            "invalid data type for byte array reader read to view type - {}",
+            data_type
+        )),
+    }
+}
+
+/// An [`ArrayReader`] for variable length byte arrays
+#[allow(unused)]
+struct ByteViewArrayReader {
+    data_type: ArrowType,
+    pages: Box<dyn PageIterator>,
+    def_levels_buffer: Option<Vec<i16>>,
+    rep_levels_buffer: Option<Vec<i16>>,
+    record_reader: GenericRecordReader<ViewBuffer, ByteViewArrayColumnValueDecoder>,
+}
+
+impl ByteViewArrayReader {
+    #[allow(unused)]
+    fn new(
+        pages: Box<dyn PageIterator>,
+        data_type: ArrowType,
+        record_reader: GenericRecordReader<ViewBuffer, ByteViewArrayColumnValueDecoder>,
+    ) -> Self {
+        Self {
+            data_type,
+            pages,
+            def_levels_buffer: None,
+            rep_levels_buffer: None,
+            record_reader,
+        }
+    }
+}
+
+impl ArrayReader for ByteViewArrayReader {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn get_data_type(&self) -> &ArrowType {
+        &self.data_type
+    }
+
+    fn read_records(&mut self, batch_size: usize) -> Result<usize> {
+        read_records(&mut self.record_reader, self.pages.as_mut(), batch_size)
+    }
+
+    fn consume_batch(&mut self) -> Result<ArrayRef> {
+        let buffer = self.record_reader.consume_record_data();
+        let null_buffer = self.record_reader.consume_bitmap_buffer();
+        self.def_levels_buffer = self.record_reader.consume_def_levels();
+        self.rep_levels_buffer = self.record_reader.consume_rep_levels();
+        self.record_reader.reset();
+
+        let array = buffer.into_array(null_buffer, &self.data_type);
+
+        Ok(array)
+    }
+
+    fn skip_records(&mut self, num_records: usize) -> Result<usize> {
+        skip_records(&mut self.record_reader, self.pages.as_mut(), num_records)
+    }
+
+    fn get_def_levels(&self) -> Option<&[i16]> {
+        self.def_levels_buffer.as_deref()
+    }
+
+    fn get_rep_levels(&self) -> Option<&[i16]> {
+        self.rep_levels_buffer.as_deref()
+    }
+}
+
+/// A [`ColumnValueDecoder`] for variable length byte arrays
+struct ByteViewArrayColumnValueDecoder {
+    dict: Option<ViewBuffer>,
+    decoder: Option<ByteViewArrayDecoder>,
+    validate_utf8: bool,
+}
+
+impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
+    type Buffer = ViewBuffer;
+
+    fn new(desc: &ColumnDescPtr) -> Self {
+        let validate_utf8 = desc.converted_type() == ConvertedType::UTF8;
+        Self {
+            dict: None,
+            decoder: None,
+            validate_utf8,
+        }
+    }
+
+    fn set_dict(
+        &mut self,
+        buf: Bytes,
+        num_values: u32,
+        encoding: Encoding,
+        _is_sorted: bool,
+    ) -> Result<()> {
+        if !matches!(
+            encoding,
+            Encoding::PLAIN | Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY
+        ) {
+            return Err(nyi_err!(
+                "Invalid/Unsupported encoding type for dictionary: {}",
+                encoding
+            ));
+        }
+
+        let mut buffer = ViewBuffer::default();
+        let mut decoder = ByteViewArrayDecoderPlain::new(
+            buf,
+            num_values as usize,
+            Some(num_values as usize),
+            self.validate_utf8,
+        );
+        decoder.read(&mut buffer, usize::MAX)?;
+        self.dict = Some(buffer);
+        Ok(())
+    }
+
+    fn set_data(
+        &mut self,
+        encoding: Encoding,
+        data: Bytes,
+        num_levels: usize,
+        num_values: Option<usize>,
+    ) -> Result<()> {
+        self.decoder = Some(ByteViewArrayDecoder::new(
+            encoding,
+            data,
+            num_levels,
+            num_values,
+            self.validate_utf8,
+        )?);
+        Ok(())
+    }
+
+    fn read(&mut self, out: &mut Self::Buffer, num_values: usize) -> Result<usize> {
+        let decoder = self
+            .decoder
+            .as_mut()
+            .ok_or_else(|| general_err!("no decoder set"))?;
+
+        decoder.read(out, num_values, self.dict.as_ref())
+    }
+
+    fn skip_values(&mut self, num_values: usize) -> Result<usize> {
+        let decoder = self
+            .decoder
+            .as_mut()
+            .ok_or_else(|| general_err!("no decoder set"))?;
+
+        decoder.skip(num_values, self.dict.as_ref())
+    }
+}
+
+/// A generic decoder from uncompressed parquet value data to [`ViewBuffer`]
+pub enum ByteViewArrayDecoder {
+    Plain(ByteViewArrayDecoderPlain),
+}
+
+impl ByteViewArrayDecoder {
+    pub fn new(
+        encoding: Encoding,
+        data: Bytes,
+        num_levels: usize,
+        num_values: Option<usize>,
+        validate_utf8: bool,
+    ) -> Result<Self> {
+        let decoder = match encoding {
+            Encoding::PLAIN => ByteViewArrayDecoder::Plain(ByteViewArrayDecoderPlain::new(
+                data,
+                num_levels,
+                num_values,
+                validate_utf8,
+            )),
+            Encoding::RLE_DICTIONARY
+            | Encoding::PLAIN_DICTIONARY
+            | Encoding::DELTA_LENGTH_BYTE_ARRAY
+            | Encoding::DELTA_BYTE_ARRAY => unimplemented!("stay tuned!"),
+            _ => {
+                return Err(general_err!(
+                    "unsupported encoding for byte array: {}",
+                    encoding
+                ))
+            }
+        };
+
+        Ok(decoder)
+    }
+
+    /// Read up to `len` values to `out` with the optional dictionary
+    pub fn read(
+        &mut self,
+        out: &mut ViewBuffer,
+        len: usize,
+        _dict: Option<&ViewBuffer>,
+    ) -> Result<usize> {
+        match self {
+            ByteViewArrayDecoder::Plain(d) => d.read(out, len),
+        }
+    }
+
+    /// Skip `len` values
+    pub fn skip(&mut self, len: usize, _dict: Option<&ViewBuffer>) -> Result<usize> {
+        match self {
+            ByteViewArrayDecoder::Plain(d) => d.skip(len),
+        }
+    }
+}
+
+/// Decoder from [`Encoding::PLAIN`] data to [`ViewBuffer`]
+pub struct ByteViewArrayDecoderPlain {
+    buf: Bytes,
+    offset: usize,
+
+    validate_utf8: bool,
+
+    /// This is a maximum as the null count is not always known, e.g. value data from
+    /// a v1 data page
+    max_remaining_values: usize,
+}
+
+impl ByteViewArrayDecoderPlain {
+    pub fn new(
+        buf: Bytes,
+        num_levels: usize,
+        num_values: Option<usize>,
+        validate_utf8: bool,
+    ) -> Self {
+        Self {
+            buf,
+            offset: 0,
+            max_remaining_values: num_values.unwrap_or(num_levels),
+            validate_utf8,
+        }
+    }
+
+    pub fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
+        let block_id = output.append_block(self.buf.clone().into());
+
+        let to_read = len.min(self.max_remaining_values);
+
+        let buf = self.buf.as_ref();
+        let mut read = 0;
+        output.views.reserve(to_read);
+        while self.offset < self.buf.len() && read != to_read {
+            if self.offset + 4 > self.buf.len() {
+                return Err(ParquetError::EOF("eof decoding byte array".into()));
+            }
+            let len_bytes: [u8; 4] = unsafe {
+                buf.get_unchecked(self.offset..self.offset + 4)
+                    .try_into()
+                    .unwrap()
+            };
+            let len = u32::from_le_bytes(len_bytes);
+
+            let start_offset = self.offset + 4;
+            let end_offset = start_offset + len as usize;
+            if end_offset > buf.len() {
+                return Err(ParquetError::EOF("eof decoding byte array".into()));
+            }
+
+            if self.validate_utf8 {
+                check_valid_utf8(unsafe { buf.get_unchecked(start_offset..end_offset) })?;
+            }
+
+            unsafe {
+                output.append_view_unchecked(block_id, start_offset as u32, len);
+            }
+            self.offset = end_offset;
+            read += 1;
+        }
+
+        self.max_remaining_values -= to_read;
+        Ok(to_read)
+    }
+
+    pub fn skip(&mut self, to_skip: usize) -> Result<usize> {
+        let to_skip = to_skip.min(self.max_remaining_values);
+        let mut skip = 0;
+        let buf = self.buf.as_ref();
+
+        while self.offset < self.buf.len() && skip != to_skip {
+            if self.offset + 4 > buf.len() {
+                return Err(ParquetError::EOF("eof decoding byte array".into()));
+            }
+            let len_bytes: [u8; 4] = buf[self.offset..self.offset + 4].try_into().unwrap();
+            let len = u32::from_le_bytes(len_bytes) as usize;
+            skip += 1;
+            self.offset = self.offset + 4 + len;
+        }
+        self.max_remaining_values -= skip;
+        Ok(skip)
+    }
+}
+
+/// Check that `val` is a valid UTF-8 sequence
+pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
+    match std::str::from_utf8(val) {
+        Ok(_) => Ok(()),
+        Err(e) => Err(general_err!("encountered non UTF-8 data: {}", e)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow_array::StringViewArray;
+    use arrow_buffer::Buffer;
+
+    use crate::{
+        arrow::{
+            array_reader::test_util::{byte_array_all_encodings, utf8_column},
+            buffer::view_buffer::ViewBuffer,
+            record_reader::buffer::ValuesBuffer,
+        },
+        basic::Encoding,
+        column::reader::decoder::ColumnValueDecoder,
+    };
+
+    use super::*;
+
+    #[test]
+    fn test_byte_array_string_view_decoder() {
+        let (pages, encoded_dictionary) =
+            byte_array_all_encodings(vec!["hello", "world", "large payload over 12 bytes", "b"]);
+
+        let column_desc = utf8_column();
+        let mut decoder = ByteViewArrayColumnValueDecoder::new(&column_desc);
+
+        decoder
+            .set_dict(encoded_dictionary, 4, Encoding::RLE_DICTIONARY, false)
+            .unwrap();
+
+        for (encoding, page) in pages {
+            if encoding != Encoding::PLAIN {
+                // skip non-plain encodings for now as they are not yet implemented
+                continue;
+            }
+            let mut output = ViewBuffer::default();
+            decoder.set_data(encoding, page, 4, Some(4)).unwrap();
+
+            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
+            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
+            assert_eq!(decoder.read(&mut output, 2).unwrap(), 2);
+            assert_eq!(decoder.read(&mut output, 4).unwrap(), 0);
+
+            assert_eq!(output.views.len(), 4);
+            assert_eq!(output.buffers.len(), 4);
+
+            let valid = [false, false, true, true, false, true, true, false, false];
+            let valid_buffer = Buffer::from_iter(valid.iter().cloned());
+
+            output.pad_nulls(0, 4, valid.len(), valid_buffer.as_slice());
+            let array = output.into_array(Some(valid_buffer), &ArrowType::Utf8View);
+            let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();
+
+            assert_eq!(
+                strings.iter().collect::<Vec<_>>(),
+                vec![
+                    None,
+                    None,
+                    Some("hello"),
+                    Some("world"),
+                    None,
+                    Some("large payload over 12 bytes"),
+                    Some("b"),
+                    None,
+                    None,
+                ]
+            );
+        }
+    }
+}
diff --git a/parquet/src/arrow/array_reader/mod.rs b/parquet/src/arrow/array_reader/mod.rs
index 8febe41e6..fef1e8722 100644
--- a/parquet/src/arrow/array_reader/mod.rs
+++ b/parquet/src/arrow/array_reader/mod.rs
@@ -32,6 +32,7 @@ use crate::file::reader::{FilePageIterator, FileReader};
 mod builder;
 mod byte_array;
 mod byte_array_dictionary;
+mod byte_view_array;
 mod empty_array;
 mod fixed_len_byte_array;
 mod fixed_size_list_array;
@@ -46,10 +47,10 @@ mod test_util;
 
 pub use builder::build_array_reader;
 pub use byte_array::make_byte_array_reader;
-#[allow(unused_imports)] // Only used for benchmarks
-pub use byte_array::make_byte_view_array_reader;
 pub use byte_array_dictionary::make_byte_array_dictionary_reader;
 #[allow(unused_imports)] // Only used for benchmarks
+pub use byte_view_array::make_byte_view_array_reader;
+#[allow(unused_imports)] // Only used for benchmarks
 pub use fixed_len_byte_array::make_fixed_len_byte_array_reader;
 pub use fixed_size_list_array::FixedSizeListArrayReader;
 pub use list_array::ListArrayReader;
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index 070dda97c..b38092dbc 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -25,25 +25,24 @@ use arrow_array::Array;
 use arrow_array::{RecordBatch, RecordBatchReader};
 use arrow_schema::{ArrowError, DataType as ArrowType, Schema, SchemaRef};
 use arrow_select::filter::prep_null_mask_filter;
+pub use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
+pub use selection::{RowSelection, RowSelector};
 
+pub use crate::arrow::array_reader::RowGroups;
 use crate::arrow::array_reader::{build_array_reader, ArrayReader};
 use crate::arrow::schema::{parquet_to_arrow_schema_and_fields, ParquetField};
-use crate::arrow::{FieldLevels, ProjectionMask};
+use crate::arrow::{parquet_to_arrow_field_levels, FieldLevels, ProjectionMask};
+use crate::column::page::{PageIterator, PageReader};
 use crate::errors::{ParquetError, Result};
+use crate::file::footer;
 use crate::file::metadata::ParquetMetaData;
+use crate::file::page_index::index_reader;
 use crate::file::reader::{ChunkReader, SerializedPageReader};
 use crate::schema::types::SchemaDescriptor;
 
 mod filter;
 mod selection;
 
-pub use crate::arrow::array_reader::RowGroups;
-use crate::column::page::{PageIterator, PageReader};
-use crate::file::footer;
-use crate::file::page_index::index_reader;
-pub use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
-pub use selection::{RowSelection, RowSelector};
-
 /// Builder for constructing parquet readers into arrow.
 ///
 /// Most users should use one of the following specializations:
@@ -250,6 +249,8 @@ impl<T> ArrowReaderBuilder<T> {
 pub struct ArrowReaderOptions {
     /// Should the reader strip any user defined metadata from the Arrow schema
     skip_arrow_metadata: bool,
+    /// If provided used as the schema for the file, otherwise the schema is read from the file
+    supplied_schema: Option<SchemaRef>,
     /// If true, attempt to read `OffsetIndex` and `ColumnIndex`
     pub(crate) page_index: bool,
 }
@@ -273,6 +274,60 @@ impl ArrowReaderOptions {
         }
     }
 
+    /// Provide a schema to use when reading the parquet file. If provided it
+    /// takes precedence over the schema inferred from the file or the schema defined
+    /// in the file's metadata. If the schema is not compatible with the file's
+    /// schema an error will be returned when constructing the builder.
+    ///
+    /// This option is only required if you want to cast columns to a different type.
+    /// For example, if you wanted to cast from an Int64 in the Parquet file to a Timestamp
+    /// in the Arrow schema.
+    ///
+    /// The supplied schema must have the same number of columns as the parquet schema and
+    /// the column names need to be the same.
+    ///
+    /// # Example
+    /// ```
+    /// use std::io::Bytes;
+    /// use std::sync::Arc;
+    /// use tempfile::tempfile;
+    /// use arrow_array::{ArrayRef, Int32Array, RecordBatch};
+    /// use arrow_schema::{DataType, Field, Schema, TimeUnit};
+    /// use parquet::arrow::arrow_reader::{ArrowReaderOptions, ParquetRecordBatchReaderBuilder};
+    /// use parquet::arrow::ArrowWriter;
+    ///
+    /// // Write data - schema is inferred from the data to be Int32
+    /// let file = tempfile().unwrap();
+    /// let batch = RecordBatch::try_from_iter(vec![
+    ///     ("col_1", Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef),
+    /// ]).unwrap();
+    /// let mut writer = ArrowWriter::try_new(file.try_clone().unwrap(), batch.schema(), None).unwrap();
+    /// writer.write(&batch).unwrap();
+    /// writer.close().unwrap();
+    ///
+    /// // Read the file back.
+    /// // Supply a schema that interprets the Int32 column as a Timestamp.
+    /// let supplied_schema = Arc::new(Schema::new(vec![
+    ///     Field::new("col_1", DataType::Timestamp(TimeUnit::Nanosecond, None), false)
+    /// ]));
+    /// let options = ArrowReaderOptions::new().with_schema(supplied_schema.clone());
+    /// let mut builder = ParquetRecordBatchReaderBuilder::try_new_with_options(
+    ///     file.try_clone().unwrap(),
+    ///     options
+    /// ).expect("Error if the schema is not compatible with the parquet file schema.");
+    ///
+    /// // Create the reader and read the data using the supplied schema.
+    /// let mut reader = builder.build().unwrap();
+    /// let _batch = reader.next().unwrap().unwrap();   
+    /// ```
+    pub fn with_schema(self, schema: SchemaRef) -> Self {
+        Self {
+            supplied_schema: Some(schema),
+            skip_arrow_metadata: true,
+            ..self
+        }
+    }
+
     /// Enable reading [`PageIndex`], if present (defaults to `false`)
     ///
     /// The `PageIndex` can be used to push down predicates to the parquet scan,
@@ -353,22 +408,77 @@ impl ArrowReaderMetadata {
     /// This function does not attempt to load the PageIndex if not present in the metadata.
     /// See [`Self::load`] for more details.
     pub fn try_new(metadata: Arc<ParquetMetaData>, options: ArrowReaderOptions) -> Result<Self> {
-        let kv_metadata = match options.skip_arrow_metadata {
-            true => None,
-            false => metadata.file_metadata().key_value_metadata(),
-        };
+        match options.supplied_schema {
+            Some(supplied_schema) => Self::with_supplied_schema(metadata, supplied_schema.clone()),
+            None => {
+                let kv_metadata = match options.skip_arrow_metadata {
+                    true => None,
+                    false => metadata.file_metadata().key_value_metadata(),
+                };
 
-        let (schema, fields) = parquet_to_arrow_schema_and_fields(
-            metadata.file_metadata().schema_descr(),
+                let (schema, fields) = parquet_to_arrow_schema_and_fields(
+                    metadata.file_metadata().schema_descr(),
+                    ProjectionMask::all(),
+                    kv_metadata,
+                )?;
+
+                Ok(Self {
+                    metadata,
+                    schema: Arc::new(schema),
+                    fields: fields.map(Arc::new),
+                })
+            }
+        }
+    }
+
+    fn with_supplied_schema(
+        metadata: Arc<ParquetMetaData>,
+        supplied_schema: SchemaRef,
+    ) -> Result<Self> {
+        let parquet_schema = metadata.file_metadata().schema_descr();
+        let field_levels = parquet_to_arrow_field_levels(
+            parquet_schema,
             ProjectionMask::all(),
-            kv_metadata,
+            Some(supplied_schema.fields()),
         )?;
-
-        Ok(Self {
-            metadata,
-            schema: Arc::new(schema),
-            fields: fields.map(Arc::new),
-        })
+        let fields = field_levels.fields;
+        let inferred_len = fields.len();
+        let supplied_len = supplied_schema.fields().len();
+        // Ensure the supplied schema has the same number of columns as the parquet schema.
+        // parquet_to_arrow_field_levels is expected to throw an error if the schemas have
+        // different lengths, but we check here to be safe.
+        if inferred_len != supplied_len {
+            Err(arrow_err!(format!(
+                "incompatible arrow schema, expected {} columns received {}",
+                inferred_len, supplied_len
+            )))
+        } else {
+            let diff_fields: Vec<_> = supplied_schema
+                .fields()
+                .iter()
+                .zip(fields.iter())
+                .filter_map(|(field1, field2)| {
+                    if field1 != field2 {
+                        Some(field1.name().clone())
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+
+            if !diff_fields.is_empty() {
+                Err(ParquetError::ArrowError(format!(
+                    "incompatible arrow schema, the following fields could not be cast: [{}]",
+                    diff_fields.join(", ")
+                )))
+            } else {
+                Ok(Self {
+                    metadata,
+                    schema: supplied_schema,
+                    fields: field_levels.levels.map(Arc::new),
+                })
+            }
+        }
     }
 
     /// Returns a reference to the [`ParquetMetaData`] for this parquet file
@@ -842,7 +952,7 @@ mod tests {
     use arrow_array::*;
     use arrow_buffer::{i256, ArrowNativeType, Buffer, IntervalDayTime};
     use arrow_data::ArrayDataBuilder;
-    use arrow_schema::{ArrowError, DataType as ArrowDataType, Field, Fields, Schema};
+    use arrow_schema::{ArrowError, DataType as ArrowDataType, Field, Fields, Schema, SchemaRef};
     use arrow_select::concat::concat_batches;
 
     use crate::arrow::arrow_reader::{
@@ -2307,10 +2417,12 @@ mod tests {
     fn test_invalid_utf8_string_array() {
         test_invalid_utf8_string_array_inner::<i32>();
     }
+
     #[test]
     fn test_invalid_utf8_large_string_array() {
         test_invalid_utf8_string_array_inner::<i64>();
     }
+
     fn test_invalid_utf8_string_array_inner<O: OffsetSizeTrait>() {
         let cases = [
             (
@@ -2620,6 +2732,275 @@ mod tests {
         assert_eq!(reader.schema(), schema_without_metadata);
     }
 
+    fn write_parquet_from_iter<I, F>(value: I) -> File
+    where
+        I: IntoIterator<Item = (F, ArrayRef)>,
+        F: AsRef<str>,
+    {
+        let batch = RecordBatch::try_from_iter(value).unwrap();
+        let file = tempfile().unwrap();
+        let mut writer =
+            ArrowWriter::try_new(file.try_clone().unwrap(), batch.schema().clone(), None).unwrap();
+        writer.write(&batch).unwrap();
+        writer.close().unwrap();
+        file
+    }
+
+    fn run_schema_test_with_error<I, F>(value: I, schema: SchemaRef, expected_error: &str)
+    where
+        I: IntoIterator<Item = (F, ArrayRef)>,
+        F: AsRef<str>,
+    {
+        let file = write_parquet_from_iter(value);
+        let options_with_schema = ArrowReaderOptions::new().with_schema(schema.clone());
+        let builder = ParquetRecordBatchReaderBuilder::try_new_with_options(
+            file.try_clone().unwrap(),
+            options_with_schema,
+        );
+        assert_eq!(builder.err().unwrap().to_string(), expected_error);
+    }
+
+    #[test]
+    fn test_schema_too_few_columns() {
+        run_schema_test_with_error(
+            vec![
+                ("int64", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+                ("int32", Arc::new(Int32Array::from(vec![0])) as ArrayRef),
+            ],
+            Arc::new(Schema::new(vec![Field::new(
+                "int64",
+                ArrowDataType::Int64,
+                false,
+            )])),
+            "Arrow: incompatible arrow schema, expected 2 struct fields got 1",
+        );
+    }
+
+    #[test]
+    fn test_schema_too_many_columns() {
+        run_schema_test_with_error(
+            vec![("int64", Arc::new(Int64Array::from(vec![0])) as ArrayRef)],
+            Arc::new(Schema::new(vec![
+                Field::new("int64", ArrowDataType::Int64, false),
+                Field::new("int32", ArrowDataType::Int32, false),
+            ])),
+            "Arrow: incompatible arrow schema, expected 1 struct fields got 2",
+        );
+    }
+
+    #[test]
+    fn test_schema_mismatched_column_names() {
+        run_schema_test_with_error(
+            vec![("int64", Arc::new(Int64Array::from(vec![0])) as ArrayRef)],
+            Arc::new(Schema::new(vec![Field::new(
+                "other",
+                ArrowDataType::Int64,
+                false,
+            )])),
+            "Arrow: incompatible arrow schema, expected field named int64 got other",
+        );
+    }
+
+    #[test]
+    fn test_schema_incompatible_columns() {
+        run_schema_test_with_error(
+            vec![
+                (
+                    "col1_invalid",
+                    Arc::new(Int64Array::from(vec![0])) as ArrayRef,
+                ),
+                (
+                    "col2_valid",
+                    Arc::new(Int32Array::from(vec![0])) as ArrayRef,
+                ),
+                (
+                    "col3_invalid",
+                    Arc::new(Date64Array::from(vec![0])) as ArrayRef,
+                ),
+            ],
+            Arc::new(Schema::new(vec![
+                Field::new("col1_invalid", ArrowDataType::Int32, false),
+                Field::new("col2_valid", ArrowDataType::Int32, false),
+                Field::new("col3_invalid", ArrowDataType::Int32, false),
+            ])),
+            "Arrow: incompatible arrow schema, the following fields could not be cast: [col1_invalid, col3_invalid]",
+        );
+    }
+
+    #[test]
+    fn test_one_incompatible_nested_column() {
+        let nested_fields = Fields::from(vec![
+            Field::new("nested1_valid", ArrowDataType::Utf8, false),
+            Field::new("nested1_invalid", ArrowDataType::Int64, false),
+        ]);
+        let nested = StructArray::try_new(
+            nested_fields,
+            vec![
+                Arc::new(StringArray::from(vec!["a"])) as ArrayRef,
+                Arc::new(Int64Array::from(vec![0])) as ArrayRef,
+            ],
+            None,
+        )
+        .expect("struct array");
+        let supplied_nested_fields = Fields::from(vec![
+            Field::new("nested1_valid", ArrowDataType::Utf8, false),
+            Field::new("nested1_invalid", ArrowDataType::Int32, false),
+        ]);
+        run_schema_test_with_error(
+            vec![
+                ("col1", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+                ("col2", Arc::new(Int32Array::from(vec![0])) as ArrayRef),
+                ("nested", Arc::new(nested) as ArrayRef),
+            ],
+            Arc::new(Schema::new(vec![
+                Field::new("col1", ArrowDataType::Int64, false),
+                Field::new("col2", ArrowDataType::Int32, false),
+                Field::new(
+                    "nested",
+                    ArrowDataType::Struct(supplied_nested_fields),
+                    false,
+                ),
+            ])),
+            "Arrow: incompatible arrow schema, the following fields could not be cast: [nested]",
+        );
+    }
+
+    #[test]
+    fn test_with_schema() {
+        let nested_fields = Fields::from(vec![
+            Field::new("utf8_to_dict", ArrowDataType::Utf8, false),
+            Field::new("int64_to_ts_nano", ArrowDataType::Int64, false),
+        ]);
+
+        let nested_arrays: Vec<ArrayRef> = vec![
+            Arc::new(StringArray::from(vec!["a", "a", "a", "b"])) as ArrayRef,
+            Arc::new(Int64Array::from(vec![1, 2, 3, 4])) as ArrayRef,
+        ];
+
+        let nested = StructArray::try_new(nested_fields, nested_arrays, None).unwrap();
+
+        let file = write_parquet_from_iter(vec![
+            (
+                "int32_to_ts_second",
+                Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+            ),
+            (
+                "date32_to_date64",
+                Arc::new(Date32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+            ),
+            ("nested", Arc::new(nested) as ArrayRef),
+        ]);
+
+        let supplied_nested_fields = Fields::from(vec![
+            Field::new(
+                "utf8_to_dict",
+                ArrowDataType::Dictionary(
+                    Box::new(ArrowDataType::Int32),
+                    Box::new(ArrowDataType::Utf8),
+                ),
+                false,
+            ),
+            Field::new(
+                "int64_to_ts_nano",
+                ArrowDataType::Timestamp(
+                    arrow::datatypes::TimeUnit::Nanosecond,
+                    Some("+10:00".into()),
+                ),
+                false,
+            ),
+        ]);
+
+        let supplied_schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "int32_to_ts_second",
+                ArrowDataType::Timestamp(arrow::datatypes::TimeUnit::Second, Some("+01:00".into())),
+                false,
+            ),
+            Field::new("date32_to_date64", ArrowDataType::Date64, false),
+            Field::new(
+                "nested",
+                ArrowDataType::Struct(supplied_nested_fields),
+                false,
+            ),
+        ]));
+
+        let options = ArrowReaderOptions::new().with_schema(supplied_schema.clone());
+        let mut arrow_reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+            file.try_clone().unwrap(),
+            options,
+        )
+        .expect("reader builder with schema")
+        .build()
+        .expect("reader with schema");
+
+        assert_eq!(arrow_reader.schema(), supplied_schema);
+        let batch = arrow_reader.next().unwrap().unwrap();
+        assert_eq!(batch.num_columns(), 3);
+        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(
+            batch
+                .column(0)
+                .as_any()
+                .downcast_ref::<TimestampSecondArray>()
+                .expect("downcast to timestamp second")
+                .value_as_datetime_with_tz(0, "+01:00".parse().unwrap())
+                .map(|v| v.to_string())
+                .expect("value as datetime"),
+            "1970-01-01 01:00:00 +01:00"
+        );
+        assert_eq!(
+            batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<Date64Array>()
+                .expect("downcast to date64")
+                .value_as_date(0)
+                .map(|v| v.to_string())
+                .expect("value as date"),
+            "1970-01-01"
+        );
+
+        let nested = batch
+            .column(2)
+            .as_any()
+            .downcast_ref::<StructArray>()
+            .expect("downcast to struct");
+
+        let nested_dict = nested
+            .column(0)
+            .as_any()
+            .downcast_ref::<Int32DictionaryArray>()
+            .expect("downcast to dictionary");
+
+        assert_eq!(
+            nested_dict
+                .values()
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .expect("downcast to string")
+                .iter()
+                .collect::<Vec<_>>(),
+            vec![Some("a"), Some("b")]
+        );
+
+        assert_eq!(
+            nested_dict.keys().iter().collect::<Vec<_>>(),
+            vec![Some(0), Some(0), Some(0), Some(1)]
+        );
+
+        assert_eq!(
+            nested
+                .column(1)
+                .as_any()
+                .downcast_ref::<TimestampNanosecondArray>()
+                .expect("downcast to timestamp nanosecond")
+                .value_as_datetime_with_tz(0, "+10:00".parse().unwrap())
+                .map(|v| v.to_string())
+                .expect("value as datetime"),
+            "1970-01-01 10:00:00.000000001 +10:00"
+        );
+    }
+
     #[test]
     fn test_empty_projection() {
         let testdata = arrow::util::test_util::parquet_test_data();


Test patch:
diff --git a/parquet_derive_test/src/lib.rs b/parquet_derive_test/src/lib.rs
index 310a3d8c9..e7c7896cb 100644
--- a/parquet_derive_test/src/lib.rs
+++ b/parquet_derive_test/src/lib.rs
@@ -248,7 +248,7 @@ mod tests {
             maybe_u64: Some(4563424),
             isize: -365,
             float: 3.5,
-            double: std::f64::NAN,
+            double: f64::NAN,
             now: chrono::Utc::now().naive_local(),
             date: chrono::naive::NaiveDate::from_ymd_opt(2015, 3, 14).unwrap(),
             uuid: uuid::Uuid::new_v4(),