Repository: apache/arrow-rs
Pull request #5939 (2024-06-22 17:25:48)

Provide Arrow Schema Hint to Parquet Reader - Alternative 2
# Which issue does this PR close?

Closes  #5657. Alternative to #5671. 

# Rationale for this change

The parquet reader automatically uses an embedded arrow schema to hint type inference for decode. In particular if the hinted type is compatible with the underlying parquet type, it performs a cast. However, in situations where the writer was not an arrow writer the schema is not available and the arrow types are inferred from the parquet schema. This is not always desirable. 

For example, you may want to cast an INT64 column to a Timestamp column or a Timestamp column to a different timezone. This PR allows a schema to be provided for use with type hinting.  

# What changes are included in this PR?

This is a draft PR for discussion. There are some API alternatives that can be considered. These are described in the section on user facing changes.

Currently, this PR adds `supplied_schema` field to `ArrowReaderOptions` and modifies the `try_new` method of ArrowReaderMetadata to use the `supplied_schema` instead of the metadata when one is provided. This option defaults to `None` so it should be backwards compatible? 

The option is provided with a new `with_schema` method on `ArrowReaderOptions`.

Adds tests showing a simple success case. Once the approach is finalized, these can be expanded to cover off all the valid cast scenarios using an approach similar to `run_single_column_read_tests`.

It also adds test showing two failure scenarios:

1. The `supplied_schema` must have exactly the same number of columns as the parquet schema or an error is thrown.
2.  The final inferred schema must be contained by the supplied schema.

# Are there any user-facing changes?

A `with_schema` method is added to the ArrowReaderOptions. It is intended to be used as follows:
```Rust
let file = File::open("file.parquet");
let schema = Arc::new(Schema::new(vec![Fields::new("col", DataType::Int32, false)]));
let options = ArrowReaderOptions::new().with_schema(schema);
// This may fail as described above.
let builder = ParquetRecordBatchReaderBuilder::try_new_with_options(file, options).unwrap();
let reader = builder.build().unwrap();
```
It need to be provided as an option before the builder is constructed so it can be utilized to provide type hints when the metadata is being read. 

There are different approaches that could be taken with the supplied_schema:
1. Currently, it needs to provide an complete schema for the entire file. It seems like it might be possible for it be treated like a partial schema that would override the the inferred schema or any schema in the metadata. This would be more ergonomic if only wanting to update a subset of fields is needed.
2. If the supplied type hints can't be utilized the API doesn't need to return an error and just use the type from the Parquet schema. This is what happens with the type hints derived from the metadata, so it might be better to make it consistent.

alamb: Thanks @efredine  -- I started the CI on this PR
efredine: I expanded the tests to include converting from an String to a Dictionary and also included a nested field. I also applied what I learnt about constructing record batches from an iterator ;-).

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/arrow-arith/src/numeric.rs b/arrow-arith/src/numeric.rs
index 17b794762..91aaf6284 100644
--- a/arrow-arith/src/numeric.rs
+++ b/arrow-arith/src/numeric.rs
@@ -888,15 +888,15 @@ mod tests {
 
         test_neg_primitive::<Int32Type>(
             &[i32::MIN],
-            Err("Compute error: Overflow happened on: -2147483648"),
+            Err("Compute error: Overflow happened on: - -2147483648"),
         );
         test_neg_primitive::<Int64Type>(
             &[i64::MIN],
-            Err("Compute error: Overflow happened on: -9223372036854775808"),
+            Err("Compute error: Overflow happened on: - -9223372036854775808"),
         );
         test_neg_primitive::<DurationSecondType>(
             &[i64::MIN],
-            Err("Compute error: Overflow happened on: -9223372036854775808"),
+            Err("Compute error: Overflow happened on: - -9223372036854775808"),
         );
 
         let r = neg_wrapping(&Int32Array::from(vec![i32::MIN])).unwrap();
@@ -911,7 +911,7 @@ mod tests {
 
         assert_eq!(
             err,
-            "Compute error: Overflow happened on: -9223372036854775808"
+            "Compute error: Overflow happened on: - -9223372036854775808"
         );
 
         let a = Decimal128Array::from(vec![1, 3, -44, 2, 4])
diff --git a/arrow-array/src/arithmetic.rs b/arrow-array/src/arithmetic.rs
index 72989ad7d..078c2e3bc 100644
--- a/arrow-array/src/arithmetic.rs
+++ b/arrow-array/src/arithmetic.rs
@@ -237,7 +237,7 @@ macro_rules! native_type_op {
             #[inline]
             fn neg_checked(self) -> Result<Self, ArrowError> {
                 self.checked_neg().ok_or_else(|| {
-                    ArrowError::ComputeError(format!("Overflow happened on: {:?}", self))
+                    ArrowError::ComputeError(format!("Overflow happened on: - {:?}", self))
                 })
             }
 
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/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/lib.rs b/object_store/src/lib.rs
index 1339d1c01..efbfe0bd4 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -550,7 +550,7 @@ pub mod integration;
 
 pub use attributes::*;
 
-pub use parse::{parse_url, parse_url_opts};
+pub use parse::{parse_url, parse_url_opts, ObjectStoreScheme};
 pub use payload::*;
 pub use upload::*;
 pub use util::{coalesce_ranges, collect_bytes, GetRange, OBJECT_STORE_COALESCE_DEFAULT};
diff --git a/object_store/src/parse.rs b/object_store/src/parse.rs
index e5d5149fb..debc9e529 100644
--- a/object_store/src/parse.rs
+++ b/object_store/src/parse.rs
@@ -24,7 +24,7 @@ use snafu::Snafu;
 use url::Url;
 
 #[derive(Debug, Snafu)]
-enum Error {
+pub enum Error {
     #[snafu(display("Unable to recognise URL \"{}\"", url))]
     Unrecognised { url: Url },
 
@@ -41,9 +41,27 @@ impl From<Error> for super::Error {
     }
 }
 
-/// Recognises various URL formats, identifying the relevant [`ObjectStore`]
-#[derive(Debug, Eq, PartialEq)]
-enum ObjectStoreScheme {
+/// Recognizes various URL formats, identifying the relevant [`ObjectStore`]
+///
+/// See [`ObjectStoreScheme::parse`] for more details
+///
+/// # Supported formats:
+/// - `file:///path/to/my/file` -> [`LocalFileSystem`]
+/// - `memory:///` -> [`InMemory`]
+/// - `s3://bucket/path` -> [`AmazonS3`](crate::aws::AmazonS3) (also supports `s3a`)
+/// - `gs://bucket/path` -> [`GoogleCloudStorage`](crate::gcp::GoogleCloudStorage)
+/// - `az://account/container/path` -> [`MicrosoftAzure`](crate::azure::MicrosoftAzure) (also supports `adl`, `azure`, `abfs`, `abfss`)
+/// - `http://mydomain/path` -> [`HttpStore`](crate::http::HttpStore)
+/// - `https://mydomain/path` -> [`HttpStore`](crate::http::HttpStore)
+///
+/// There are also special cases for AWS and Azure for `https://{host?}/path` paths:
+/// - `dfs.core.windows.net`, `blob.core.windows.net`, `dfs.fabric.microsoft.com`, `blob.fabric.microsoft.com` -> [`MicrosoftAzure`](crate::azure::MicrosoftAzure)
+/// - `amazonaws.com` -> [`AmazonS3`](crate::aws::AmazonS3)
+/// - `r2.cloudflarestorage.com` -> [`AmazonS3`](crate::aws::AmazonS3)
+///
+#[non_exhaustive] // permit new variants
+#[derive(Debug, Eq, PartialEq, Clone)]
+pub enum ObjectStoreScheme {
     /// Url corresponding to [`LocalFileSystem`]
     Local,
     /// Url corresponding to [`InMemory`]
@@ -62,7 +80,27 @@ impl ObjectStoreScheme {
     /// Create an [`ObjectStoreScheme`] from the provided [`Url`]
     ///
     /// Returns the [`ObjectStoreScheme`] and the remaining [`Path`]
-    fn parse(url: &Url) -> Result<(Self, Path), Error> {
+    ///
+    /// # Example
+    /// ```
+    /// # use url::Url;
+    /// # use object_store::ObjectStoreScheme;
+    /// let url: Url = "file:///path/to/my/file".parse().unwrap();
+    /// let (scheme, path) = ObjectStoreScheme::parse(&url).unwrap();
+    /// assert_eq!(scheme, ObjectStoreScheme::Local);
+    /// assert_eq!(path.as_ref(), "path/to/my/file");
+    ///
+    /// let url: Url = "https://blob.core.windows.net/path/to/my/file".parse().unwrap();
+    /// let (scheme, path) = ObjectStoreScheme::parse(&url).unwrap();
+    /// assert_eq!(scheme, ObjectStoreScheme::MicrosoftAzure);
+    /// assert_eq!(path.as_ref(), "path/to/my/file");
+    ///
+    /// let url: Url = "https://example.com/path/to/my/file".parse().unwrap();
+    /// let (scheme, path) = ObjectStoreScheme::parse(&url).unwrap();
+    /// assert_eq!(scheme, ObjectStoreScheme::Http);
+    /// assert_eq!(path.as_ref(), "path/to/my/file");
+    /// ```
+    pub fn parse(url: &Url) -> Result<(Self, Path), Error> {
         let strip_bucket = || Some(url.path().strip_prefix('/')?.split_once('/')?.1);
 
         let (scheme, path) = match (url.scheme(), url.host_str()) {
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();
diff --git a/parquet_derive/src/parquet_field.rs b/parquet_derive/src/parquet_field.rs
index 9fff76c42..f99ea3e03 100644
--- a/parquet_derive/src/parquet_field.rs
+++ b/parquet_derive/src/parquet_field.rs
@@ -239,7 +239,8 @@ impl Field {
     /// because this parsing logic is not sophisticated enough for definition
     /// levels beyond 2.
     ///
-    /// `Option` types and references not supported
+    /// `Option` types and references not supported, but the column itself can be nullable
+    /// (i.e., def_level==1), as long as the values are all valid.
     pub fn reader_snippet(&self) -> proc_macro2::TokenStream {
         let ident = &self.ident;
         let column_reader = self.ty.column_reader();
@@ -248,7 +249,13 @@ impl Field {
         let write_batch_expr = quote! {
             let mut vals = Vec::new();
             if let #column_reader(mut typed) = column_reader {
-                typed.read_records(num_records, None, None, &mut vals)?;
+                let mut definition_levels = Vec::new();
+                let (total_num, valid_num, decoded_num) = typed.read_records(
+                    num_records, Some(&mut definition_levels), None, &mut vals)?;
+                if valid_num != decoded_num {
+                    panic!("Support only valid records, found {} null records in column type {}",
+                        decoded_num - valid_num, stringify!{#ident});
+                }
             } else {
                 panic!("Schema and struct disagree on type for {}", stringify!{#ident});
             }
@@ -876,15 +883,21 @@ mod test {
             snippet,
             (quote! {
                  {
-                     let mut vals = Vec::new();
-                     if let ColumnReader::Int64ColumnReader(mut typed) = column_reader {
-                         typed.read_records(num_records, None, None, &mut vals)?;
-                     } else {
-                         panic!("Schema and struct disagree on type for {}", stringify!{ counter });
-                     }
-                     for (i, r) in &mut records[..num_records].iter_mut().enumerate() {
-                         r.counter = vals[i] as usize;
-                     }
+                    let mut vals = Vec::new();
+                    if let ColumnReader::Int64ColumnReader(mut typed) = column_reader {
+                        let mut definition_levels = Vec::new();
+                        let (total_num, valid_num, decoded_num) = typed.read_records(
+                            num_records, Some(&mut definition_levels), None, &mut vals)?;
+                        if valid_num != decoded_num {
+                            panic!("Support only valid records, found {} null records in column type {}",
+                                decoded_num - valid_num, stringify!{counter});
+                        }
+                    } else {
+                        panic!("Schema and struct disagree on type for {}", stringify!{counter});
+                    }
+                    for (i, r) in &mut records[..num_records].iter_mut().enumerate() {
+                        r.counter = vals[i] as usize;
+                    }
                  }
             })
             .to_string()
@@ -1291,7 +1304,13 @@ mod test {
             {
                 let mut vals = Vec::new();
                 if let ColumnReader::Int64ColumnReader(mut typed) = column_reader {
-                    typed.read_records(num_records, None, None, &mut vals)?;
+                    let mut definition_levels = Vec::new();
+                    let (total_num, valid_num, decoded_num) = typed.read_records(
+                        num_records, Some(&mut definition_levels), None, &mut vals)?;
+                    if valid_num != decoded_num {
+                        panic!("Support only valid records, found {} null records in column type {}",
+                            decoded_num - valid_num, stringify!{henceforth});
+                    }
                 } else {
                     panic!("Schema and struct disagree on type for {}", stringify!{ henceforth });
                 }
@@ -1359,7 +1378,13 @@ mod test {
             {
                 let mut vals = Vec::new();
                 if let ColumnReader::Int32ColumnReader(mut typed) = column_reader {
-                    typed.read_records(num_records, None, None, &mut vals)?;
+                    let mut definition_levels = Vec::new();
+                    let (total_num, valid_num, decoded_num) = typed.read_records(
+                        num_records, Some(&mut definition_levels), None, &mut vals)?;
+                    if valid_num != decoded_num {
+                        panic!("Support only valid records, found {} null records in column type {}",
+                            decoded_num - valid_num, stringify!{henceforth});
+                    }
                 } else {
                     panic!("Schema and struct disagree on type for {}", stringify!{ henceforth });
                 }
@@ -1427,7 +1452,13 @@ mod test {
             {
                 let mut vals = Vec::new();
                 if let ColumnReader::FixedLenByteArrayColumnReader(mut typed) = column_reader {
-                    typed.read_records(num_records, None, None, &mut vals)?;
+                    let mut definition_levels = Vec::new();
+                    let (total_num, valid_num, decoded_num) = typed.read_records(
+                        num_records, Some(&mut definition_levels), None, &mut vals)?;
+                    if valid_num != decoded_num {
+                        panic!("Support only valid records, found {} null records in column type {}",
+                            decoded_num - valid_num, stringify!{unique_id});
+                    }
                 } else {
                     panic!("Schema and struct disagree on type for {}", stringify!{ unique_id });
                 }


Test patch:
diff --git a/parquet_derive_test/src/lib.rs b/parquet_derive_test/src/lib.rs
index b8e6c27f5..e7c7896cb 100644
--- a/parquet_derive_test/src/lib.rs
+++ b/parquet_derive_test/src/lib.rs
@@ -66,6 +66,25 @@ struct APartiallyCompleteRecord {
     pub byte_vec: Vec<u8>,
 }
 
+// This struct has OPTIONAL columns
+// If these fields are guaranteed to be valid
+// we can load this struct into APartiallyCompleteRecord
+#[derive(PartialEq, ParquetRecordWriter, Debug)]
+struct APartiallyOptionalRecord {
+    pub bool: bool,
+    pub string: String,
+    pub maybe_i16: Option<i16>,
+    pub maybe_i32: Option<i32>,
+    pub maybe_u64: Option<u64>,
+    pub isize: isize,
+    pub float: f32,
+    pub double: f64,
+    pub now: chrono::NaiveDateTime,
+    pub date: chrono::NaiveDate,
+    pub uuid: uuid::Uuid,
+    pub byte_vec: Vec<u8>,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -218,6 +237,47 @@ mod tests {
         assert_eq!(drs[0], out[0]);
     }
 
+    #[test]
+    fn test_parquet_derive_read_optional_but_valid_column() {
+        let file = get_temp_file("test_parquet_derive_read_optional", &[]);
+        let drs: Vec<APartiallyOptionalRecord> = vec![APartiallyOptionalRecord {
+            bool: true,
+            string: "a string".into(),
+            maybe_i16: Some(-45),
+            maybe_i32: Some(456),
+            maybe_u64: Some(4563424),
+            isize: -365,
+            float: 3.5,
+            double: f64::NAN,
+            now: chrono::Utc::now().naive_local(),
+            date: chrono::naive::NaiveDate::from_ymd_opt(2015, 3, 14).unwrap(),
+            uuid: uuid::Uuid::new_v4(),
+            byte_vec: vec![0x65, 0x66, 0x67],
+        }];
+
+        let generated_schema = drs.as_slice().schema().unwrap();
+
+        let props = Default::default();
+        let mut writer =
+            SerializedFileWriter::new(file.try_clone().unwrap(), generated_schema, props).unwrap();
+
+        let mut row_group = writer.next_row_group().unwrap();
+        drs.as_slice().write_to_row_group(&mut row_group).unwrap();
+        row_group.close().unwrap();
+        writer.close().unwrap();
+
+        use parquet::file::{reader::FileReader, serialized_reader::SerializedFileReader};
+        let reader = SerializedFileReader::new(file).unwrap();
+        let mut out: Vec<APartiallyCompleteRecord> = Vec::new();
+
+        let mut row_group = reader.get_row_group(0).unwrap();
+        out.read_from_row_group(&mut *row_group, 1).unwrap();
+
+        assert_eq!(drs[0].maybe_i16.unwrap(), out[0].i16);
+        assert_eq!(drs[0].maybe_i32.unwrap(), out[0].i32);
+        assert_eq!(drs[0].maybe_u64.unwrap(), out[0].u64);
+    }
+
     /// Returns file handle for a temp file in 'target' directory with a provided content
     pub fn get_temp_file(file_name: &str, content: &[u8]) -> fs::File {
         // build tmp path to a file in "target/debug/testdata"