fix: validate parquet.coerce_int96 when it is set (#25628)

## Which issue does this PR close?

- Part of #17498. This covers
`datafusion.execution.parquet.coerce_int96` only.

## Rationale for this change

The option is stored as a string, so a bad value is accepted and only
fails later:

```sql
-- Succeeds, and SHOW reports 'nanos'.
SET datafusion.execution.parquet.coerce_int96 = 'nanos';

-- Fails here, on the first table that touches Parquet:
-- Invalid or Unsupported Configuration: Unknown or unsupported parquet coerce_int96: nanos. Valid values are: ns, us, ms, and s.
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'data.parquet';
```

The same thing happens through `CREATE EXTERNAL TABLE ... OPTIONS
('format.coerce_int96' 'nanos')`. If the table has an explicit schema
the string is parsed again in `ParquetSource` with `unwrap()`, so that
path panics instead.

#20253 fixed this the same way and was reviewed, but went stale and was
closed. This picks it up with the review applied.

## What changes are included in this PR?

- `DFTimeUnit` in `parquet_config.rs`: `ns`, `us`, `ms`, `s`, parsed
case-insensitively and displayed as those strings, so `SHOW`,
`information_schema` and the proto encoding are unchanged. The
`ConfigField` impls follow `DFParquetStatistics`, so a rejected value
leaves the previous one in place and an unset option stays unset.
- `ParquetOptions.coerce_int96` becomes `Option<DFTimeUnit>`, and
`ParquetFormat::coerce_int96()` / `with_coerce_int96()` use the typed
value.
- `parse_coerce_int96_string` in `datasource-parquet` is removed, along
with the `unwrap()` that called it. `From<DFTimeUnit> for
arrow::datatypes::TimeUnit` replaces it.
- Proto conversion parses on the way in, so a bad value in a serialized
plan is an error rather than a panic.
- The option's docstring said "If true, ..." as if it were a boolean. It
now says what the value is and lists the valid values. `configs.md` and
`information_schema.slt` carry that text.

I kept the `Option`. #20253 looked at dropping it, but `None` means "do
not coerce" and `coerce_int96_tz` only applies when `coerce_int96` is
set, so `None` and `Some(ns)` are not the same.

## What is the testing strategy for this PR?

- `set_variable.slt`: set, show, one rejected value with its full error
text, and that the setting is unchanged afterwards.
- `test_parquet_coerce_int96_validation` in `config.rs`: valid values, a
rejected value keeps the previous one, an unset value stays unset,
nested keys are rejected.
- `time_unit_parses_and_displays` in `parquet_config.rs`.
- `test_invalid_parquet_coerce_int96_rejected_from_proto` in
`proto-common`, and the two existing round-trip tests now use the typed
value.
- I also checked `CREATE EXTERNAL TABLE ... OPTIONS
('format.coerce_int96' 'nanos')` fails at CREATE time with the new
message. The existing `coerce_int96` data tests in `parquet.slt` and the
core Parquet tests pass unchanged.

## Are there any user-facing changes?

Yes. An invalid `datafusion.execution.parquet.coerce_int96` is rejected
by `SET` and by table options with a message, instead of being accepted
and failing (or panicking) at scan time.

This changes the public field `ParquetOptions::coerce_int96` from
`Option<String>` to `Option<DFTimeUnit>`, and the two `ParquetFormat`
accessors with it, so it needs the `api change` label. SQL and
`ConfigOptions::set` users are not affected.

I searched GitHub for code that assigns the field directly. These will
need a one-line change on upgrade, from `Some("us".to_string())` to
`Some(DFTimeUnit::Microsecond)` or `Some("us".parse()?)`:

- apache/datafusion-comet: `native/core/src/parquet/parquet_exec.rs`
hardcodes `"us"` for Spark, plus one test assertion.
- delta-io/delta-rs: `crates/core/src/datafile/properties.rs` uses
`get_or_insert_with(|| "us".to_string())`.
- lakehq/sail: `crates/sail-session/src/session_factory/mod.rs` assigns
`"us"`, and `crates/sail-data-source/src/formats/parquet/options.rs`
copies the field through its own `Option<String>`.
- DoumanAsh/mishka: `src/datafusion.rs` assigns an owned string.

Code that already uses `DFParquetMetadata::with_coerce_int96` or the
opener (apache/auron does) takes an arrow `TimeUnit` and is unaffected.

I can add a note to the 56.0.0 upgrade guide matching the
`ExecutionOptions::time_zone` entry from #25788 if that's wanted.

Author: akashjainn

datafusion/common/src/config.rs

@@ -24,7 +24,7 @@ use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
 use crate::error::{_config_datafusion_err, _config_err};
 use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
 use crate::parquet_config::{
-    DFParquetCompression, DFParquetStatistics, DFParquetWriterVersion,
+    DFParquetCompression, DFParquetStatistics, DFParquetWriterVersion, DFTimeUnit,
     RowGroupRangeAssignment,
 };
 use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
@@ -1403,13 +1403,13 @@ config_namespace! {
         /// reading them as binary and then casting to string.
         pub binary_as_string: bool, default = false
 
-        /// (reading) If true, parquet reader will read columns of
-        /// physical type int96 as originating from a different resolution
-        /// than nanosecond. This is useful for reading data from systems like Spark
-        /// which stores microsecond resolution timestamps in an int96 allowing it
-        /// to write values with a larger date range than 64-bit timestamps with
-        /// nanosecond resolution.
-        pub coerce_int96: Option<String>, transform = str::to_lowercase, default = None
+        /// (reading) If set, parquet reader will read columns of
+        /// physical type int96 as originating from this resolution instead of
+        /// nanosecond. Valid values are: ns, us, ms, and s. This is useful for
+        /// reading data from systems like Spark which store microsecond
+        /// resolution timestamps in an int96, allowing them to write values with
+        /// a larger date range than 64-bit timestamps with nanosecond resolution.
+        pub coerce_int96: Option<DFTimeUnit>, default = None
 
         /// (reading) Optional timezone applied to INT96 columns when `coerce_int96`
         /// is set. When `Some`, INT96 columns coerce to
@@ -5019,6 +5019,70 @@ mod tests {
         assert_eq!(scalar, DFParquetCompression::Lz4);
     }
 
+    #[cfg(feature = "parquet")]
+    #[test]
+    fn test_parquet_coerce_int96_validation() {
+        use crate::{config::ConfigOptions, parquet_config::DFTimeUnit};
+
+        let mut config = ConfigOptions::default();
+        assert_eq!(config.execution.parquet.coerce_int96, None);
+
+        for (value, expected) in [
+            ("ns", DFTimeUnit::Nanosecond),
+            ("US", DFTimeUnit::Microsecond),
+            ("ms", DFTimeUnit::Millisecond),
+            ("s", DFTimeUnit::Second),
+        ] {
+            config
+                .set("datafusion.execution.parquet.coerce_int96", value)
+                .unwrap();
+            assert_eq!(config.execution.parquet.coerce_int96, Some(expected));
+        }
+
+        let err = config
+            .set("datafusion.execution.parquet.coerce_int96", "nanos")
+            .unwrap_err();
+        assert_contains!(
+            err.to_string(),
+            "Invalid parquet coerce_int96 setting: nanos. Expected one of: ns, us, ms, s"
+        );
+        // A rejected value leaves the previous one in place.
+        assert_eq!(
+            config.execution.parquet.coerce_int96,
+            Some(DFTimeUnit::Second)
+        );
+
+        // An invalid update must leave an unset option unset rather than
+        // inserting a default.
+        config.execution.parquet.coerce_int96 = None;
+        assert!(
+            config
+                .set("datafusion.execution.parquet.coerce_int96", "nanos")
+                .is_err()
+        );
+        assert_eq!(config.execution.parquet.coerce_int96, None);
+
+        config.execution.parquet.coerce_int96 = Some(DFTimeUnit::Microsecond);
+        assert!(
+            config
+                .set("datafusion.execution.parquet.coerce_int96.typo", "ns")
+                .is_err()
+        );
+        assert!(
+            config
+                .reset("datafusion.execution.parquet.coerce_int96.typo")
+                .is_err()
+        );
+        assert_eq!(
+            config.execution.parquet.coerce_int96,
+            Some(DFTimeUnit::Microsecond)
+        );
+
+        let mut scalar = DFTimeUnit::Microsecond;
+        assert!(ConfigField::set(&mut scalar, "typo", "ns").is_err());
+        assert_eq!(scalar, DFTimeUnit::Microsecond);
+    }
+
     #[cfg(feature = "parquet")]
     #[test]
     fn set_cdc_enabled_flag() {

datafusion/common/src/parquet_config.rs

@@ -481,6 +481,110 @@ impl From<DFParquetCompression> for parquet::basic::Compression {
     }
 }
 
+/// Time unit for reading Parquet INT96 timestamps
+///
+/// This enum validates `coerce_int96` at configuration time, so only `ns`,
+/// `us`, `ms` or `s` can be set via `SET` commands or deserialization.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DFTimeUnit {
+    Nanosecond,
+    Microsecond,
+    Millisecond,
+    Second,
+}
+
+impl FromStr for DFTimeUnit {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "ns" => Ok(Self::Nanosecond),
+            "us" => Ok(Self::Microsecond),
+            "ms" => Ok(Self::Millisecond),
+            "s" => Ok(Self::Second),
+            other => Err(DataFusionError::Configuration(format!(
+                "Invalid parquet coerce_int96 setting: {other}. Expected one of: ns, us, ms, s"
+            ))),
+        }
+    }
+}
+
+impl Display for DFTimeUnit {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let s = match self {
+            Self::Nanosecond => "ns",
+            Self::Microsecond => "us",
+            Self::Millisecond => "ms",
+            Self::Second => "s",
+        };
+        f.write_str(s)
+    }
+}
+
+impl ConfigField for DFTimeUnit {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, key: &str, value: &str) -> Result<()> {
+        if !key.is_empty() {
+            return crate::error::_config_err!(
+                "Config field parquet.coerce_int96 is a scalar DFTimeUnit and does not have nested field \"{}\"",
+                key
+            );
+        }
+
+        *self = Self::from_str(value)?;
+        Ok(())
+    }
+}
+
+/// `ConfigField` for `Option<DFTimeUnit>` parses before assigning so an
+/// invalid value does not turn an unset option into the default.
+impl ConfigField for Option<DFTimeUnit> {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        match self {
+            Some(time_unit) => time_unit.visit(v, key, description),
+            None => v.none(key, description),
+        }
+    }
+
+    fn set(&mut self, key: &str, value: &str) -> Result<()> {
+        if !key.is_empty() {
+            return crate::error::_config_err!(
+                "Config field parquet.coerce_int96 is a scalar Option<DFTimeUnit> and does not have nested field \"{}\"",
+                key
+            );
+        }
+
+        *self = Some(DFTimeUnit::from_str(value)?);
+        Ok(())
+    }
+
+    fn reset(&mut self, key: &str) -> Result<()> {
+        if key.is_empty() {
+            *self = None;
+            Ok(())
+        } else {
+            crate::error::_config_err!(
+                "Config field parquet.coerce_int96 is a scalar Option<DFTimeUnit> and does not have nested field \"{}\"",
+                key
+            )
+        }
+    }
+}
+
+impl From<DFTimeUnit> for arrow::datatypes::TimeUnit {
+    fn from(value: DFTimeUnit) -> Self {
+        match value {
+            DFTimeUnit::Nanosecond => Self::Nanosecond,
+            DFTimeUnit::Microsecond => Self::Microsecond,
+            DFTimeUnit::Millisecond => Self::Millisecond,
+            DFTimeUnit::Second => Self::Second,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -566,4 +670,25 @@ mod tests {
         }
         assert!(ZstdLevel::try_new(*ZSTD_LEVELS.end() as i32 + 1).is_err());
     }
+
+    #[test]
+    fn time_unit_parses_and_displays() {
+        for (input, expected, displayed) in [
+            ("ns", DFTimeUnit::Nanosecond, "ns"),
+            ("US", DFTimeUnit::Microsecond, "us"),
+            ("ms", DFTimeUnit::Millisecond, "ms"),
+            ("s", DFTimeUnit::Second, "s"),
+        ] {
+            let parsed: DFTimeUnit = input.parse().unwrap();
+            assert_eq!(parsed, expected, "input: {input}");
+            assert_eq!(parsed.to_string(), displayed);
+        }
+        for input in ["nanos", "", "ns "] {
+            let err = input.parse::<DFTimeUnit>().unwrap_err().to_string();
+            assert!(
+                err.contains("Invalid parquet coerce_int96 setting"),
+                "input {input:?}: {err}"
+            );
+        }
+    }
 }

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -47,6 +47,7 @@ mod tests {
     use arrow_schema::{SchemaRef, TimeUnit};
     use bytes::{BufMut, BytesMut};
     use datafusion_common::config::TableParquetOptions;
+    use datafusion_common::parquet_config::DFTimeUnit;
     use datafusion_common::test_util::{batches_to_sort_string, batches_to_string};
     use datafusion_common::{Result, ScalarValue, assert_contains};
     use datafusion_datasource::file_format::FileFormat;
@@ -1355,7 +1356,7 @@ mod tests {
                 ])),
             ),
             (
-                Some("ns".to_string()),
+                Some(DFTimeUnit::Nanosecond),
                 Arc::new(Int64Array::from(vec![
                     Some(1704141296123456000),
                     Some(1704070800000000000),
@@ -1366,7 +1367,7 @@ mod tests {
                 ])),
             ),
             (
-                Some("us".to_string()),
+                Some(DFTimeUnit::Microsecond),
                 Arc::new(Int64Array::from(vec![
                     Some(1704141296123456),
                     Some(1704070800000000),
@@ -1381,7 +1382,7 @@ mod tests {
         for (time_unit, expected) in time_units_and_expected {
             let parquet_exec = scan_format(
                 &state,
-                &ParquetFormat::default().with_coerce_int96(time_unit.clone()),
+                &ParquetFormat::default().with_coerce_int96(time_unit),
                 Some(schema.clone()),
                 &testdata,
                 filename,
@@ -1430,7 +1431,7 @@ mod tests {
 
         let parquet_exec = scan_format(
             &state,
-            &ParquetFormat::default().with_coerce_int96(Some("us".to_string())),
+            &ParquetFormat::default().with_coerce_int96(Some(DFTimeUnit::Microsecond)),
             None,
             testdata,
             filename,

datafusion/datasource-parquet/src/file_format.rs

@@ -45,6 +45,7 @@ use datafusion_datasource::file_format::{
 use datafusion_common::Statistics;
 use datafusion_common::config::{ConfigField, ConfigFileType, TableParquetOptions};
 use datafusion_common::encryption::FileDecryptionProperties;
+use datafusion_common::parquet_config::DFTimeUnit;
 use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_common::{
     DEFAULT_PARQUET_EXTENSION, DataFusionError, GetExt, Result, internal_datafusion_err,
@@ -61,9 +62,7 @@ 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, parse_coerce_int96_tz_string,
-};
+use crate::source::{ParquetSource, parse_coerce_int96_tz_string};
 use async_trait::async_trait;
 use bytes::Bytes;
 use datafusion_datasource::source::DataSourceExec;
@@ -242,14 +241,14 @@ impl ParquetFormat {
     /// Get [`coerce_int96`]
     ///
     /// [`coerce_int96`]: datafusion_common::config::ParquetOptions::coerce_int96
-    pub fn coerce_int96(&self) -> Option<String> {
-        self.options.global.coerce_int96.clone()
+    pub fn coerce_int96(&self) -> Option<DFTimeUnit> {
+        self.options.global.coerce_int96
     }
 
     /// Set [`coerce_int96`]
     ///
     /// [`coerce_int96`]: datafusion_common::config::ParquetOptions::coerce_int96
-    pub fn with_coerce_int96(mut self, time_unit: Option<String>) -> Self {
+    pub fn with_coerce_int96(mut self, time_unit: Option<DFTimeUnit>) -> Self {
         self.options.global.coerce_int96 = time_unit;
         self
     }
@@ -361,10 +360,7 @@ impl FileFormat for ParquetFormat {
         store: &Arc<dyn ObjectStore>,
         objects: &[ObjectMeta],
     ) -> Result<SchemaRef> {
-        let coerce_int96 = match self.coerce_int96() {
-            Some(time_unit) => Some(parse_coerce_int96_string(time_unit.as_str())?),
-            None => None,
-        };
+        let coerce_int96 = self.coerce_int96().map(arrow::datatypes::TimeUnit::from);
         let coerce_int96_tz = self
             .options
             .global
@@ -762,8 +758,8 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
             write_row_group_number_distinct_values: global_options
                 .global
                 .write_row_group_number_distinct_values,
-            coerce_int96_opt: global_options.global.coerce_int96.map(|compression| {
-                parquet_options::CoerceInt96Opt::CoerceInt96(compression)
+            coerce_int96_opt: global_options.global.coerce_int96.map(|time_unit| {
+                parquet_options::CoerceInt96Opt::CoerceInt96(time_unit.to_string())
             }),
             coerce_int96_tz_opt: global_options.global.coerce_int96_tz.map(|tz| {
                 parquet_options::CoerceInt96TzOpt::CoerceInt96Tz(tz)

datafusion/datasource-parquet/src/source.rs

@@ -537,24 +537,6 @@ impl ParquetSource {
     }
 }
 
-/// Parses datafusion.common.config.ParquetOptions.coerce_int96 String to a arrow_schema.datatype.TimeUnit
-pub(crate) fn parse_coerce_int96_string(
-    str_setting: &str,
-) -> datafusion_common::Result<TimeUnit> {
-    let str_setting_lower: &str = &str_setting.to_lowercase();
-
-    match str_setting_lower {
-        "ns" => Ok(TimeUnit::Nanosecond),
-        "us" => Ok(TimeUnit::Microsecond),
-        "ms" => Ok(TimeUnit::Millisecond),
-        "s" => Ok(TimeUnit::Second),
-        _ => Err(DataFusionError::Configuration(format!(
-            "Unknown or unsupported parquet coerce_int96: \
-        {str_setting}. Valid values are: ns, us, ms, and s."
-        ))),
-    }
-}
-
 /// Validates that `tz` is a parseable IANA timezone and returns it as an
 /// `Arc<str>` for use in `Timestamp(_, Some(tz))` types.
 pub(crate) fn parse_coerce_int96_tz_string(
@@ -617,8 +599,7 @@ impl FileSource for ParquetSource {
             .table_parquet_options
             .global
             .coerce_int96
-            .as_ref()
-            .map(|time_unit| parse_coerce_int96_string(time_unit.as_str()).unwrap());
+            .map(TimeUnit::from);
         let coerce_int96_tz = self
             .table_parquet_options
             .global

datafusion/ffi/src/session/mod.rs

@@ -926,7 +926,8 @@ mod tests {
         table_options.json.schema_infer_max_rec = Some(10);
         #[cfg(feature = "parquet")]
         {
-            table_options.parquet.global.coerce_int96 = Some("123456789".into());
+            table_options.parquet.global.coerce_int96 =
+                Some(datafusion_common::parquet_config::DFTimeUnit::Microsecond);
         }
         table_options.current_format = Some(ConfigFileType::JSON);