Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
116 changes: 114 additions & 2 deletions datafusion/common/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -943,7 +943,7 @@ config_namespace! {
/// The default time zone
///
/// Some functions, e.g. `now` return timestamps in this time zone
pub time_zone: Option<String>, default = None
pub time_zone: Option<ConfigTimeZone>, default = None

/// Parquet options
pub parquet: ParquetOptions, default = Default::default()
Expand Down Expand Up @@ -2013,6 +2013,76 @@ impl From<ConfigDurationFormat> for arrow::util::display::DurationFormat {
}
}

/// A time zone that is known to parse as an Arrow
/// [`Tz`](arrow::array::timezone::Tz), used for
/// [`ExecutionOptions::time_zone`].
///
/// The value is kept as it was written, so `+08` stays `+08` rather than
/// being normalized to `+08:00`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfigTimeZone(String);

impl ConfigTimeZone {
/// Returns the time zone as it was written.
pub fn as_str(&self) -> &str {
&self.0
}
}

impl FromStr for ConfigTimeZone {
type Err = DataFusionError;

fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.parse::<arrow::array::timezone::Tz>() {
Ok(_) => Ok(Self(s.to_string())),
Err(_) => _config_err!(
"Invalid time zone: {s}. Valid values are UTC offsets such as +08:00 or IANA time zone names such as Asia/Taipei"
),
}
}
}

impl Display for ConfigTimeZone {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}

/// `ConfigField` for `Option<ConfigTimeZone>` parses before assigning so an
/// invalid value leaves the current setting unchanged.
impl ConfigField for Option<ConfigTimeZone> {
fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
match self {
Some(time_zone) => v.some(key, time_zone, description),
None => v.none(key, description),
}
}

fn set(&mut self, key: &str, value: &str) -> Result<()> {
if !key.is_empty() {
return _config_err!(
"Config field is a scalar Option<ConfigTimeZone> and does not have nested field \"{}\"",
key
);
}

*self = Some(ConfigTimeZone::from_str(value)?);
Ok(())
}

fn reset(&mut self, key: &str) -> Result<()> {
if key.is_empty() {
*self = None;
Ok(())
} else {
_config_err!(
"Config field is a scalar Option<ConfigTimeZone> and does not have nested field \"{}\"",
key
)
}
}
}

config_namespace! {
/// Options controlling the format of output when printing record batches
/// Copies [`arrow::util::display::FormatOptions`]
Expand Down Expand Up @@ -4061,7 +4131,6 @@ impl Display for OutputFormat {

#[cfg(test)]
mod tests {
#[cfg(feature = "parquet")]
use crate::assert_contains;
use crate::config::TableParquetOptions;
use crate::config::{
Expand Down Expand Up @@ -4694,6 +4763,49 @@ mod tests {
assert_eq!(scalar, DFParquetStatistics::Page);
}

#[test]
fn test_execution_time_zone_validation() {
use crate::config::ConfigOptions;

const KEY: &str = "datafusion.execution.time_zone";
let mut config = ConfigOptions::default();
assert_eq!(config.execution.time_zone, None);

// Valid values are kept exactly as written.
for value in ["+08:00", "-08:00", "+0800", "+08", "UTC", "Asia/Taipei"] {
config.set(KEY, value).unwrap();
assert_eq!(
config.execution.time_zone.as_ref().map(|tz| tz.as_str()),
Some(value)
);
}

// A rejected value leaves the previous one in place.
for value in ["+08:00:00", "08:00", "08", "Asia/Taipei2", "AEST", ""] {
let err = config.set(KEY, value).unwrap_err();
assert_contains!(err.to_string(), format!("Invalid time zone: {value}."));
assert_eq!(
config.execution.time_zone.as_ref().map(|tz| tz.as_str()),
Some("Asia/Taipei")
);
}

// An invalid update of an unset value leaves it unset.
config.reset(KEY).unwrap();
assert_eq!(config.execution.time_zone, None);
assert!(config.set(KEY, "Asia/Taipei2").is_err());
assert_eq!(config.execution.time_zone, None);

// The option has no nested fields.
config.set(KEY, "UTC").unwrap();
assert!(config.set(&format!("{KEY}.typo"), "UTC").is_err());
assert!(config.reset(&format!("{KEY}.typo")).is_err());
assert_eq!(
config.execution.time_zone.as_ref().map(|tz| tz.as_str()),
Some("UTC")
);
}

#[cfg(feature = "parquet")]
#[test]
fn test_parquet_compression_validation() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1965,7 +1965,12 @@ async fn test_config_options_work_for_scalar_func() -> Result<()> {
}

fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
let tz = args.config_options.execution.time_zone.clone();
let tz = args
.config_options
.execution
.time_zone
.as_ref()
.map(ToString::to_string);
Ok(ColumnarValue::Scalar(ScalarValue::from(tz)))
}
}
Expand All @@ -1975,7 +1980,7 @@ async fn test_config_options_work_for_scalar_func() -> Result<()> {
});

let mut config = SessionConfig::new();
config.options_mut().execution.time_zone = Some("AEST".into());
config.options_mut().execution.time_zone = Some("Australia/Sydney".parse().unwrap());

let ctx = SessionContext::new_with_config(config);

Expand All @@ -1988,7 +1993,7 @@ async fn test_config_options_work_for_scalar_func() -> Result<()> {
let expected_schema = Schema::new(vec![Field::new("a", DataType::Utf8, false)]);
let expected = RecordBatch::try_new(
SchemaRef::from(expected_schema),
vec![create_array!(Utf8, ["AEST"])],
vec![create_array!(Utf8, ["Australia/Sydney"])],
)?;

assert_eq!(expected, actual[0]);
Expand Down
7 changes: 6 additions & 1 deletion datafusion/ffi/src/tests/udf_udaf_udwf.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,12 @@ impl ScalarUDFImpl for TimeZoneUDF {
&self,
args: ScalarFunctionArgs,
) -> datafusion_common::Result<ColumnarValue> {
let tz = args.config_options.execution.time_zone.clone();
let tz = args
.config_options
.execution
.time_zone
.as_ref()
.map(ToString::to_string);
Ok(ColumnarValue::Scalar(ScalarValue::from(tz)))
}

Expand Down
7 changes: 4 additions & 3 deletions datafusion/ffi/tests/ffi_udf.rs
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,8 @@ mod tests {
assert!(result[0].column(0).as_string::<i32>().is_null(0));

let mut config = SessionConfig::new();
config.options_mut().execution.time_zone = Some("AEST".into());
config.options_mut().execution.time_zone =
Some("Australia/Sydney".parse().unwrap());

let ctx = SessionContext::new_with_config(config);

Expand All @@ -154,7 +155,7 @@ mod tests {
assert_eq!(result.len(), 1);
assert!(!result[0].column(0).as_string::<i32>().is_null(0));
let result = result[0].column(0).as_string::<i32>().value(0);
assert_eq!(result, "AEST");
assert_eq!(result, "Australia/Sydney");

Ok(())
}
Expand All @@ -175,7 +176,7 @@ mod tests {
);

let mut options = ConfigOptions::default();
options.execution.time_zone = Some("AEST".into());
options.execution.time_zone = Some("Australia/Sydney".parse().unwrap());

let updated = foreign_udf
.with_updated_config(&options)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,7 @@ fn criterion_benchmark(c: &mut Criterion) {
let arg_field = Field::new("a", DataType::Utf8, false).into();
let arg_fields = vec![arg_field];
let mut options = ConfigOptions::default();
options.execution.time_zone = Some("UTC".into());
options.execution.time_zone = Some("UTC".parse().unwrap());
let config_options = Arc::new(options);

let to_timestamp_udf = to_timestamp(config_options.as_ref());
Expand Down
2 changes: 1 addition & 1 deletion datafusion/functions/src/datetime/current_date.rs
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ impl ScalarUDFImpl for CurrentDateFunc {
.execution
.time_zone
.as_ref()
.and_then(|tz| tz.parse::<Tz>().ok())
.and_then(|tz| tz.as_str().parse::<Tz>().ok())
.map_or_else(
|| datetime_to_days(&now_ts),
|tz| {
Expand Down
4 changes: 2 additions & 2 deletions datafusion/functions/src/datetime/current_time.rs
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ impl ScalarUDFImpl for CurrentTimeFunc {
.execution
.time_zone
.as_ref()
.and_then(|tz| tz.parse::<Tz>().ok())
.and_then(|tz| tz.as_str().parse::<Tz>().ok())
.map_or_else(
|| datetime_to_time_nanos(&now_ts),
|tz| {
Expand Down Expand Up @@ -158,7 +158,7 @@ mod tests {
config.execution.time_zone = if tz.is_empty() {
None
} else {
Some(tz.to_string())
Some(tz.parse().unwrap())
};
let schema = Arc::new(DFSchema::empty());
SimplifyContext::builder()
Expand Down
6 changes: 3 additions & 3 deletions datafusion/functions/src/datetime/from_unixtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -353,7 +353,7 @@ mod test {
timezone: Option<&str>,
) -> Result<ColumnarValue> {
let mut options = ConfigOptions::default();
options.execution.time_zone = timezone.map(str::to_string);
options.execution.time_zone = timezone.map(|tz| tz.parse().unwrap());
let func = FromUnixtimeFunc::new_with_config(&options);

let arg_field: FieldRef = Field::new("a", DataType::Int64, true).into();
Expand Down Expand Up @@ -560,7 +560,7 @@ mod test {
#[test]
fn test_session_timezone_is_used_without_explicit_timezone() {
let mut options = ConfigOptions::default();
options.execution.time_zone = Some("America/Denver".to_string());
options.execution.time_zone = Some("America/Denver".parse().unwrap());

let func = FromUnixtimeFunc::new_with_config(&options);

Expand Down Expand Up @@ -599,7 +599,7 @@ mod test {
#[test]
fn test_explicit_timezone_overrides_session_timezone() {
let mut options = ConfigOptions::default();
options.execution.time_zone = Some("America/Denver".to_string());
options.execution.time_zone = Some("America/Denver".parse().unwrap());

let func = FromUnixtimeFunc::new_with_config(&options);

Expand Down
Loading
Loading