feat: optional timezone for coerce_int96

Add a new TableParquetOptions field coerce_int96_tz that, when set,
makes coerce_int96_to_resolution produce Timestamp(unit, Some(tz))
instead of the default Timestamp(unit, None). Threads the new option
through ParquetSource, ParquetOpener, and DFParquetMetadata.

Spark and other systems write INT96 as UTC-adjusted instants, so
downstream readers that need the resulting Arrow type to carry the
LTZ signal (for example, Comet's schema adapter rejecting INT96 ->
TimestampNTZ on Spark 3.x) can set this to "UTC". Default behavior is
unchanged when the option is unset.

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;

datafusion/datasource-parquet/src/file_format.rs

@@ -713,10 +713,17 @@ pub fn apply_file_schema_type_coercions(
 }
 
 /// Coerces the file schema's Timestamps to the provided TimeUnit if Parquet schema contains INT96.
+///
+/// When `timezone` is `Some`, INT96-derived columns coerce to
+/// `Timestamp(time_unit, Some(timezone))`; otherwise they coerce to
+/// `Timestamp(time_unit, None)` (the historical default). Spark and other
+/// systems write INT96 as UTC-adjusted instants, so callers that need the
+/// resulting Arrow type to be timezone-aware should pass `Some("UTC".into())`.
 pub fn coerce_int96_to_resolution(
     parquet_schema: &SchemaDescriptor,
     file_schema: &Schema,
     time_unit: &TimeUnit,
+    timezone: Option<Arc<str>>,
 ) -> Option<Schema> {
     // Traverse the parquet_schema columns looking for int96 physical types. If encountered, insert
     // the field's full path into a set.
@@ -879,11 +886,11 @@ pub fn coerce_int96_to_resolution(
                 (DataType::Timestamp(TimeUnit::Nanosecond, None), None)
                     if int96_fields.contains(parquet_path.concat().as_str()) =>
                 // We found a timestamp(nanos) and it originated as int96. Coerce it to the correct
-                // time_unit.
+                // time_unit, optionally attaching the requested timezone.
                 {
                     parent_fields.borrow_mut().push(field_with_new_type(
                         current_field,
-                        DataType::Timestamp(*time_unit, None),
+                        DataType::Timestamp(*time_unit, timezone.clone()),
                     ));
                 }
                 // Other types can be cloned as they are.
@@ -1852,7 +1859,7 @@ mod tests {
         let arrow_schema = parquet_to_arrow_schema(&descr, None).unwrap();
 
         let result =
-            coerce_int96_to_resolution(&descr, &arrow_schema, &TimeUnit::Microsecond)
+            coerce_int96_to_resolution(&descr, &arrow_schema, &TimeUnit::Microsecond, None)
                 .unwrap();
 
         // Only the first field (c0) should be converted to a microsecond timestamp because it's the
@@ -1939,7 +1946,7 @@ mod tests {
         let arrow_schema = parquet_to_arrow_schema(&descr, None).unwrap();
 
         let result =
-            coerce_int96_to_resolution(&descr, &arrow_schema, &TimeUnit::Microsecond)
+            coerce_int96_to_resolution(&descr, &arrow_schema, &TimeUnit::Microsecond, None)
                 .unwrap();
 
         let expected_schema = Schema::new(vec![

datafusion/datasource-parquet/src/metadata.rs

@@ -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
@@ -222,6 +231,7 @@ impl<'a> DFParquetMetadata<'a> {
                     file_metadata.schema_descr(),
                     &schema,
                     time_unit,
+                    self.coerce_int96_tz.clone(),
                 )
             })
             .unwrap_or(schema);

datafusion/datasource-parquet/src/opener.rs

@@ -119,6 +119,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>>,
@@ -281,6 +285,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>,
@@ -651,6 +656,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,
@@ -782,6 +788,7 @@ impl MetadataLoadedParquetOpen {
                 reader_metadata.parquet_schema(),
                 &physical_file_schema,
                 coerce,
+                prepared.coerce_int96_tz.clone(),
             )
         {
             physical_file_schema = Arc::new(merged);
@@ -1748,6 +1755,7 @@ 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,
+                coerce_int96_tz: None,
                 #[cfg(feature = "parquet_encryption")]
                 file_decryption_properties: None,
                 expr_adapter_factory: Arc::new(DefaultPhysicalExprAdapterFactory),

datafusion/datasource-parquet/src/source.rs

@@ -557,6 +557,12 @@ impl FileSource for ParquetSource {
             .coerce_int96
             .as_ref()
             .map(|time_unit| parse_coerce_int96_string(time_unit.as_str()).unwrap());
+        let coerce_int96_tz = self
+            .table_parquet_options
+            .global
+            .coerce_int96_tz
+            .as_ref()
+            .map(|tz| Arc::<str>::from(tz.as_str()));
 
         Ok(Box::new(ParquetMorselizer {
             partition_index: partition,
@@ -578,6 +584,7 @@ impl FileSource for ParquetSource {
             enable_bloom_filter: self.bloom_filter_on_read(),
             enable_row_group_stats_pruning: self.table_parquet_options.global.pruning,
             coerce_int96,
+            coerce_int96_tz,
             #[cfg(feature = "parquet_encryption")]
             file_decryption_properties,
             expr_adapter_factory,