You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/16 14:05:08 UTC
(datafusion) branch gh-readonly-queue/main/pr-25628-09fb3f41f1e986b1ccbf2a0075aaff24d1ace01d created (now 9f4b005e6e)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25628-09fb3f41f1e986b1ccbf2a0075aaff24d1ace01d
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 9f4b005e6e fix: validate parquet.coerce_int96 when it is set (#25628)
This branch includes the following new commits:
new 9f4b005e6e fix: validate parquet.coerce_int96 when it is set (#25628)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: validate parquet.coerce_int96 when it is set (#25628)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25628-09fb3f41f1e986b1ccbf2a0075aaff24d1ace01d
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 9f4b005e6e89a820e7c4d014d1ba1d1ca128c5a2
Author: akashjainn <80...@users.noreply.github.com>
AuthorDate: Fri Oct 16 14:04:57 2026 +0000
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.
---
datafusion/common/src/config.rs | 80 +++++++++++--
datafusion/common/src/parquet_config.rs | 125 +++++++++++++++++++++
.../core/src/datasource/physical_plan/parquet.rs | 9 +-
datafusion/datasource-parquet/src/file_format.rs | 20 ++--
datafusion/datasource-parquet/src/source.rs | 21 +---
datafusion/ffi/src/session/mod.rs | 3 +-
datafusion/proto-common/src/from_proto/mod.rs | 36 ++++--
datafusion/proto-common/src/to_proto/mod.rs | 2 +-
datafusion/proto-models/src/from_proto.rs | 14 ++-
.../sqllogictest/test_files/information_schema.slt | 2 +-
.../sqllogictest/test_files/set_variable.slt | 37 ++++++
docs/source/user-guide/configs.md | 2 +-
12 files changed, 290 insertions(+), 61 deletions(-)
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index aebbfb9154..b2cfdbfc2e 100644
--- a/datafusion/common/src/config.rs
+++ b/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() {
diff --git a/datafusion/common/src/parquet_config.rs b/datafusion/common/src/parquet_config.rs
index 62570abd6d..cc466ed14e 100644
--- a/datafusion/common/src/parquet_config.rs
+++ b/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}"
+ );
+ }
+ }
}
diff --git a/datafusion/core/src/datasource/physical_plan/parquet.rs b/datafusion/core/src/datasource/physical_plan/parquet.rs
index d562bbe849..e0b186406b 100644
--- a/datafusion/core/src/datasource/physical_plan/parquet.rs
+++ b/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,
diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs
index 95ba19a975..ca5a4bddec 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/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)
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
index da7d295847..2a8b5d428f 100644
--- a/datafusion/datasource-parquet/src/source.rs
+++ b/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
diff --git a/datafusion/ffi/src/session/mod.rs b/datafusion/ffi/src/session/mod.rs
index fcf546f438..fcf01d4264 100644
--- a/datafusion/ffi/src/session/mod.rs
+++ b/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);
diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs
index 8e10c74c76..42b7c4976c 100644
--- a/datafusion/proto-common/src/from_proto/mod.rs
+++ b/datafusion/proto-common/src/from_proto/mod.rs
@@ -1283,9 +1283,9 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions {
)?,
schema_force_view_types: value.schema_force_view_types,
binary_as_string: value.binary_as_string,
- coerce_int96: value.coerce_int96_opt.clone().map(|opt| match opt {
- protobuf::parquet_options::CoerceInt96Opt::CoerceInt96(v) => Some(v),
- }).unwrap_or(None),
+ coerce_int96: value.coerce_int96_opt.as_ref().map(|opt| match opt {
+ protobuf::parquet_options::CoerceInt96Opt::CoerceInt96(v) => v.parse(),
+ }).transpose()?,
coerce_int96_tz: value.coerce_int96_tz_opt.clone().map(|opt| match opt {
protobuf::parquet_options::CoerceInt96TzOpt::CoerceInt96Tz(v) => Some(v),
}).unwrap_or(None),
@@ -1454,7 +1454,7 @@ mod tests {
MaxRowGroupBytes, ParquetCdcOptions, ParquetOptions, TableParquetOptions,
};
use datafusion_common::parquet_config::{
- DFParquetCompression, DFParquetStatistics, RowGroupRangeAssignment,
+ DFParquetCompression, DFParquetStatistics, DFTimeUnit, RowGroupRangeAssignment,
};
#[test]
@@ -1671,12 +1671,12 @@ mod tests {
#[test]
fn test_parquet_options_coerce_int96_tz_set_round_trip() {
let opts = ParquetOptions {
- coerce_int96: Some("us".to_string()),
+ coerce_int96: Some(DFTimeUnit::Microsecond),
coerce_int96_tz: Some("UTC".to_string()),
..ParquetOptions::default()
};
let recovered = parquet_options_proto_round_trip(opts.clone());
- assert_eq!(recovered.coerce_int96, Some("us".to_string()));
+ assert_eq!(recovered.coerce_int96, Some(DFTimeUnit::Microsecond));
assert_eq!(recovered.coerce_int96_tz, Some("UTC".to_string()));
}
@@ -1740,6 +1740,26 @@ mod tests {
);
}
+ #[test]
+ fn test_invalid_parquet_coerce_int96_rejected_from_proto() {
+ let opts = ParquetOptions::default();
+ let mut proto: crate::protobuf_common::ParquetOptions =
+ (&opts).try_into().expect("to_proto");
+ proto.coerce_int96_opt = Some(
+ crate::protobuf_common::parquet_options::CoerceInt96Opt::CoerceInt96(
+ "nanos".to_string(),
+ ),
+ );
+
+ let err = ParquetOptions::try_from(&proto).unwrap_err();
+ assert!(
+ err.to_string().contains(
+ "Invalid parquet coerce_int96 setting: nanos. Expected one of: ns, us, ms, s"
+ ),
+ "unexpected error: {err}"
+ );
+ }
+
#[test]
fn test_invalid_parquet_statistics_rejected_from_proto() {
let opts = ParquetOptions::default();
@@ -1778,10 +1798,10 @@ mod tests {
#[test]
fn test_table_parquet_options_coerce_int96_tz_round_trip() {
let mut opts = TableParquetOptions::default();
- opts.global.coerce_int96 = Some("us".to_string());
+ opts.global.coerce_int96 = Some(DFTimeUnit::Microsecond);
opts.global.coerce_int96_tz = Some("America/Los_Angeles".to_string());
let recovered = table_parquet_options_proto_round_trip(opts.clone());
- assert_eq!(recovered.global.coerce_int96, Some("us".to_string()));
+ assert_eq!(recovered.global.coerce_int96, Some(DFTimeUnit::Microsecond));
assert_eq!(
recovered.global.coerce_int96_tz,
Some("America/Los_Angeles".to_string())
diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs
index b10e65a7ac..01f37a478f 100644
--- a/datafusion/proto-common/src/to_proto/mod.rs
+++ b/datafusion/proto-common/src/to_proto/mod.rs
@@ -967,7 +967,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions {
binary_as_string: value.binary_as_string,
skip_arrow_metadata: value.skip_arrow_metadata,
write_row_group_number_distinct_values: value.write_row_group_number_distinct_values,
- coerce_int96_opt: value.coerce_int96.clone().map(protobuf::parquet_options::CoerceInt96Opt::CoerceInt96),
+ coerce_int96_opt: value.coerce_int96.map(|v| protobuf::parquet_options::CoerceInt96Opt::CoerceInt96(v.to_string())),
coerce_int96_tz_opt: value.coerce_int96_tz.clone().map(protobuf::parquet_options::CoerceInt96TzOpt::CoerceInt96Tz),
max_predicate_cache_size_opt: value.max_predicate_cache_size.map(|v| protobuf::parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(v as u64)),
max_row_group_bytes_opt: value.max_row_group_bytes.map(|v| protobuf::parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(v.get() as u64)),
diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs
index 9af35f50e4..cdf7c30d00 100644
--- a/datafusion/proto-models/src/from_proto.rs
+++ b/datafusion/proto-models/src/from_proto.rs
@@ -472,11 +472,15 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions {
skip_arrow_metadata: proto.skip_arrow_metadata,
write_row_group_number_distinct_values: proto
.write_row_group_number_distinct_values,
- coerce_int96: proto.coerce_int96_opt.as_ref().map(|opt| match opt {
- parquet_options::CoerceInt96Opt::CoerceInt96(coerce_int96) => {
- coerce_int96.clone()
- }
- }),
+ coerce_int96: proto
+ .coerce_int96_opt
+ .as_ref()
+ .map(|opt| match opt {
+ parquet_options::CoerceInt96Opt::CoerceInt96(coerce_int96) => {
+ coerce_int96.parse()
+ }
+ })
+ .transpose()?,
coerce_int96_tz: proto
.coerce_int96_tz_opt
.as_ref()
diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt
index 3780b68fe0..2381fc14e3 100644
--- a/datafusion/sqllogictest/test_files/information_schema.slt
+++ b/datafusion/sqllogictest/test_files/information_schema.slt
@@ -410,7 +410,7 @@ datafusion.execution.parquet.bloom_filter_fpp NULL (writing) Sets bloom filter f
datafusion.execution.parquet.bloom_filter_ndv NULL (writing) Sets bloom filter number of distinct values. If NULL, uses default parquet writer setting
datafusion.execution.parquet.bloom_filter_on_read true (reading) Use any available bloom filters when reading parquet files
datafusion.execution.parquet.bloom_filter_on_write false (writing) Write bloom filters for all columns when creating parquet files
-datafusion.execution.parquet.coerce_int96 NULL (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.
+datafusion.execution.parquet.coerce_int96 NULL (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.
datafusion.execution.parquet.coerce_int96_tz NULL (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 `coe [...]
datafusion.execution.parquet.column_index_truncate_length 64 (writing) Sets column index truncate length
datafusion.execution.parquet.compression zstd(3) (writing) Sets default parquet compression codec. Valid values are: uncompressed, snappy, gzip(level), brotli(level), lz4, zstd(level), and lz4_raw. These values are not case sensitive. If NULL, uses default parquet writer setting Note that this default setting is not the same as the default parquet writer setting.
diff --git a/datafusion/sqllogictest/test_files/set_variable.slt b/datafusion/sqllogictest/test_files/set_variable.slt
index ab86363f67..2ea93fcefa 100644
--- a/datafusion/sqllogictest/test_files/set_variable.slt
+++ b/datafusion/sqllogictest/test_files/set_variable.slt
@@ -800,6 +800,43 @@ datafusion.execution.parquet.compression gzip(6)
statement ok
RESET datafusion.execution.parquet.compression
+# parquet.coerce_int96 is validated when it is set
+statement ok
+SET datafusion.execution.parquet.coerce_int96 = 'US'
+
+query TT
+SHOW datafusion.execution.parquet.coerce_int96
+----
+datafusion.execution.parquet.coerce_int96 us
+
+statement error
+SET datafusion.execution.parquet.coerce_int96 = 'nanos'
+----
+DataFusion error: Error setting config datafusion.execution.parquet.coerce_int96
+caused by
+Invalid or Unsupported Configuration: Invalid parquet coerce_int96 setting: nanos. Expected one of: ns, us, ms, s
+
+
+# The rejected value did not change the setting
+query TT
+SHOW datafusion.execution.parquet.coerce_int96
+----
+datafusion.execution.parquet.coerce_int96 us
+
+statement ok
+RESET datafusion.execution.parquet.coerce_int96
+
+# Table options are validated at CREATE time too. With an explicit schema this
+# used to be accepted and then panic on the first scan.
+statement error
+CREATE EXTERNAL TABLE int96_bad_unit (a TIMESTAMP)
+STORED AS PARQUET
+LOCATION '../../parquet-testing/data/int96_from_spark.parquet'
+OPTIONS ('format.coerce_int96' 'nanos');
+----
+DataFusion error: Invalid or Unsupported Configuration: Invalid parquet coerce_int96 setting: nanos. Expected one of: ns, us, ms, s
+
+
statement error DataFusion error: Error setting config datafusion\.execution\.batch_size\ncaused by\nInvalid or Unsupported Configuration: value must be greater than 0
SET datafusion.execution.batch_size = 0
diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md
index 4a6418410f..273b0e39cf 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -93,7 +93,7 @@ The following configuration settings are available:
| datafusion.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. [...]
| datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. The parquet reader is optimized for reading `Utf8View` and `BinaryView`, so such queries are significantly faster than reading `Utf8`/`Binary` and then casting to the view types. [...]
| datafusion.execution.parquet.binary_as_string | false | (reading) If true, parquet reader will read columns of `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`. Parquet files generated by some legacy writers do not correctly set the UTF8 flag for strings, causing string columns to be loaded as BLOB instead. The parquet reader has special optimizations for `Utf8` validation, so reading such columns as strings is significant [...]
-| datafusion.execution.parquet.coerce_int96 | NULL | (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. [...]
+| datafusion.execution.parquet.coerce_int96 | NULL | (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 resolutio [...]
| datafusion.execution.parquet.coerce_int96_tz | NULL | (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` sema [...]
| datafusion.execution.parquet.bloom_filter_on_read | true | (reading) Use any available bloom filters when reading parquet files [...]
| datafusion.execution.parquet.max_predicate_cache_size | NULL | (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. [...]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org