feat: optional timezone for coerce_int96 (#22318)

## Which issue does this PR close?

N/A

## Rationale for this change

`coerce_int96_to_resolution` currently produces `Timestamp(unit, None)`
for every INT96-derived column. Some downstream readers need the
resulting Arrow type to carry a timezone, because the *absence* of a
timezone is itself meaningful.

The motivating case is Apache DataFusion Comet (a Spark accelerator)
trying to enforce [SPARK-36182\: pre-Spark-4 Spark rejects reading a
Parquet TimestampLTZ column as
TimestampNTZ](https://issues.apache.org/jira/browse/SPARK-36182).
Comet's schema adapter pattern-matches `Timestamp(_, Some(_)) ->
Timestamp(_, None)` to detect this case, but for INT96 columns the
post-coerce type is `Timestamp(unit, None)` — indistinguishable from a
true TimestampNTZ source. The LTZ signal is destroyed at the wrong
layer.

Spark and other systems write INT96 as UTC-adjusted instants, so a
caller can ask for the column to surface as `Timestamp(unit,
Some(\"UTC\"))`, preserving the LTZ semantic at the Arrow level.

## What changes are included in this PR?

- New `TableParquetOptions.global.coerce_int96_tz: Option<String>`
config field (defaults to `None`).
- `coerce_int96_to_resolution` gains a `timezone: Option<Arc<str>>`
parameter and threads it into the constructed `Timestamp` type.
- The new option is plumbed through `ParquetSource` -> `ParquetOpener` /
`ParquetMorselizer` -> `DFParquetMetadata`.
- `with_coerce_int96_tz` builder method on `DFParquetMetadata`.
- Default behavior is unchanged when the option is unset.

## Are these changes tested?

Yes, see apache/datafusion-comet#4357

## Are there any user-facing changes?

A new \`coerce_int96_tz\` config option. No change in behavior for the
default value.

---------

Co-authored-by: Oleks V <comphead@users.noreply.github.com>

Author: andygrove

datafusion/common/src/config.rs

@@ -908,6 +908,16 @@ config_namespace! {
         /// nanosecond resolution.
         pub coerce_int96: Option<String>, transform = str::to_lowercase, default = None
 
+        /// (reading) Optional timezone applied to INT96 columns when `coerce_int96`
+        /// is set. When `Some`, INT96 columns coerce to
+        /// `Timestamp(<coerce_int96>, Some(<tz>))` instead of the default
+        /// `Timestamp(<coerce_int96>, None)`. Spark and other systems write INT96
+        /// values as UTC-adjusted instants, so callers that need the resulting
+        /// Arrow type to be timezone-aware (e.g. for Spark `TimestampType`
+        /// semantics) should set this to `"UTC"`. No effect when `coerce_int96`
+        /// is `None`.
+        pub coerce_int96_tz: Option<String>, default = None
+
         /// (reading) Use any available bloom filters when reading parquet files
         pub bloom_filter_on_read: bool, default = true
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -208,6 +208,7 @@ impl ParquetOptions {
             schema_force_view_types: _,
             binary_as_string: _, // not used for writer props
             coerce_int96: _,     // not used for writer props
+            coerce_int96_tz: _,  // not used for writer props
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
         } = self;
@@ -482,6 +483,7 @@ mod tests {
             binary_as_string: defaults.binary_as_string,
             skip_arrow_metadata: defaults.skip_arrow_metadata,
             coerce_int96: None,
+            coerce_int96_tz: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
             use_content_defined_chunking: defaults.use_content_defined_chunking.clone(),
         }
@@ -600,6 +602,7 @@ mod tests {
                 binary_as_string: global_options_defaults.binary_as_string,
                 skip_arrow_metadata: global_options_defaults.skip_arrow_metadata,
                 coerce_int96: None,
+                coerce_int96_tz: None,
                 use_content_defined_chunking: props.content_defined_chunking().map(|c| {
                     CdcOptions {
                         min_chunk_size: c.min_chunk_size,

datafusion/datasource-parquet/src/file_format.rs

@@ -25,7 +25,7 @@ use std::sync::Arc;
 // Re-export so the historical `file_format::*` paths still resolve.
 #[expect(deprecated)]
 pub use crate::schema_coercion::{
-    apply_file_schema_type_coercions, coerce_file_schema_to_string_type,
+    Int96Coercer, apply_file_schema_type_coercions, coerce_file_schema_to_string_type,
     coerce_file_schema_to_view_type, coerce_int96_to_resolution,
     transform_binary_to_string, transform_schema_to_view,
 };
@@ -56,7 +56,9 @@ use datafusion_session::Session;
 
 use crate::metadata::{DFParquetMetadata, lex_ordering_to_sorting_columns};
 use crate::reader::CachedParquetFileReaderFactory;
-use crate::source::{ParquetSource, parse_coerce_int96_string};
+use crate::source::{
+    ParquetSource, parse_coerce_int96_string, parse_coerce_int96_tz_string,
+};
 use async_trait::async_trait;
 use bytes::Bytes;
 use datafusion_datasource::source::DataSourceExec;
@@ -333,6 +335,13 @@ impl FileFormat for ParquetFormat {
             Some(time_unit) => Some(parse_coerce_int96_string(time_unit.as_str())?),
             None => None,
         };
+        let coerce_int96_tz = self
+            .options
+            .global
+            .coerce_int96_tz
+            .as_ref()
+            .map(|tz| parse_coerce_int96_tz_string(tz))
+            .transpose()?;
 
         let file_metadata_cache =
             state.runtime_env().cache_manager.get_file_metadata_cache();
@@ -350,6 +359,7 @@ impl FileFormat for ParquetFormat {
                     .with_decryption_properties(file_decryption_properties)
                     .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache)))
                     .with_coerce_int96(coerce_int96)
+                    .with_coerce_int96_tz(coerce_int96_tz.clone())
                     .fetch_schema_with_location()
                     .await?;
                 Ok::<_, DataFusionError>(result)

datafusion/datasource-parquet/src/metadata.rs

@@ -18,7 +18,7 @@
 //! [`DFParquetMetadata`] for fetching Parquet file metadata, statistics
 //! and schema information.
 
-use crate::{apply_file_schema_type_coercions, coerce_int96_to_resolution};
+use crate::{Int96Coercer, apply_file_schema_type_coercions};
 use arrow::array::{Array, ArrayRef, BooleanArray};
 use arrow::compute::and;
 use arrow::compute::kernels::cmp::eq;
@@ -72,6 +72,8 @@ pub struct DFParquetMetadata<'a> {
     file_metadata_cache: Option<Arc<dyn FileMetadataCache>>,
     /// timeunit to coerce INT96 timestamps to
     pub coerce_int96: Option<TimeUnit>,
+    /// Optional timezone applied to INT96-coerced timestamps.
+    pub coerce_int96_tz: Option<Arc<str>>,
 }
 
 impl<'a> DFParquetMetadata<'a> {
@@ -83,6 +85,7 @@ impl<'a> DFParquetMetadata<'a> {
             decryption_properties: None,
             file_metadata_cache: None,
             coerce_int96: None,
+            coerce_int96_tz: None,
         }
     }
 
@@ -116,6 +119,12 @@ impl<'a> DFParquetMetadata<'a> {
         self
     }
 
+    /// Set the optional timezone applied to INT96-coerced timestamps.
+    pub fn with_coerce_int96_tz(mut self, timezone: Option<Arc<str>>) -> Self {
+        self.coerce_int96_tz = timezone;
+        self
+    }
+
     /// Fetch parquet metadata from the remote object store
     pub async fn fetch_metadata(&self) -> Result<Arc<ParquetMetaData>> {
         // implementation to fetch parquet metadata
@@ -218,11 +227,9 @@ impl<'a> DFParquetMetadata<'a> {
             .coerce_int96
             .as_ref()
             .and_then(|time_unit| {
-                coerce_int96_to_resolution(
-                    file_metadata.schema_descr(),
-                    &schema,
-                    time_unit,
-                )
+                Int96Coercer::new(file_metadata.schema_descr(), &schema, time_unit)
+                    .with_timezone(self.coerce_int96_tz.clone())
+                    .coerce()
             })
             .unwrap_or(schema);
         Ok(schema)

datafusion/datasource-parquet/src/mod.rs

@@ -54,7 +54,7 @@ pub use row_filter::can_expr_be_pushed_down_with_schemas;
 pub use row_group_filter::RowGroupAccessPlanFilter;
 #[expect(deprecated)]
 pub use schema_coercion::{
-    apply_file_schema_type_coercions, coerce_file_schema_to_string_type,
+    Int96Coercer, apply_file_schema_type_coercions, coerce_file_schema_to_string_type,
     coerce_file_schema_to_view_type, coerce_int96_to_resolution,
     transform_binary_to_string, transform_schema_to_view,
 };

datafusion/datasource-parquet/src/opener/mod.rs

@@ -29,8 +29,8 @@ use crate::push_decoder::{DecoderBuilderConfig, PushDecoderStreamState};
 use crate::row_filter::{RowFilterGenerator, build_projection_read_plan};
 use crate::row_group_filter::{BloomFilterStatistics, RowGroupAccessPlanFilter};
 use crate::{
-    ParquetAccessPlan, ParquetFileMetrics, ParquetFileReaderFactory,
-    apply_file_schema_type_coercions, coerce_int96_to_resolution,
+    Int96Coercer, ParquetAccessPlan, ParquetFileMetrics, ParquetFileReaderFactory,
+    apply_file_schema_type_coercions,
 };
 use arrow::array::RecordBatch;
 use arrow::datatypes::DataType;
@@ -121,6 +121,10 @@ pub(super) struct ParquetMorselizer {
     pub enable_row_group_stats_pruning: bool,
     /// Coerce INT96 timestamps to specific TimeUnit
     pub coerce_int96: Option<TimeUnit>,
+    /// Optional timezone applied to INT96-coerced timestamps. When `Some`, the
+    /// coerced column type becomes `Timestamp(<coerce_int96>, Some(<tz>))`.
+    /// No effect when `coerce_int96` is `None`.
+    pub coerce_int96_tz: Option<Arc<str>>,
     /// Optional parquet FileDecryptionProperties
     #[cfg(feature = "parquet_encryption")]
     pub file_decryption_properties: Option<Arc<FileDecryptionProperties>>,
@@ -283,6 +287,7 @@ struct PreparedParquetOpen {
     enable_row_group_stats_pruning: bool,
     limit: Option<usize>,
     coerce_int96: Option<TimeUnit>,
+    coerce_int96_tz: Option<Arc<str>>,
     expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,
     predicate_creation_errors: Count,
     max_predicate_cache_size: Option<usize>,
@@ -653,6 +658,7 @@ impl ParquetMorselizer {
             enable_row_group_stats_pruning: self.enable_row_group_stats_pruning,
             limit: self.limit,
             coerce_int96: self.coerce_int96,
+            coerce_int96_tz: self.coerce_int96_tz.clone(),
             expr_adapter_factory: Arc::clone(&self.expr_adapter_factory),
             predicate_creation_errors,
             max_predicate_cache_size: self.max_predicate_cache_size,
@@ -780,11 +786,13 @@ impl MetadataLoadedParquetOpen {
         }
 
         if let Some(ref coerce) = prepared.coerce_int96
-            && let Some(merged) = coerce_int96_to_resolution(
+            && let Some(merged) = Int96Coercer::new(
                 reader_metadata.parquet_schema(),
                 &physical_file_schema,
                 coerce,
             )
+            .with_timezone(prepared.coerce_int96_tz.clone())
+            .coerce()
         {
             physical_file_schema = Arc::new(merged);
             options = options.with_schema(Arc::clone(&physical_file_schema));
@@ -1603,6 +1611,10 @@ mod test {
                 enable_bloom_filter: self.enable_bloom_filter,
                 enable_row_group_stats_pruning: self.enable_row_group_stats_pruning,
                 coerce_int96: self.coerce_int96,
+                // End-to-end coercion behavior (including timezone) is
+                // covered by parquet.slt. No opener-level test currently
+                // needs a non-default value here.
+                coerce_int96_tz: None,
                 #[cfg(feature = "parquet_encryption")]
                 file_decryption_properties: None,
                 expr_adapter_factory: Arc::new(DefaultPhysicalExprAdapterFactory),