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/09/16 13:50:24 UTC
(datafusion) branch gh-readonly-queue/main/pr-25058-ddad4a821aee49586fd213ab9a4a77bb2826a076 created (now 2cb6b54085)
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-25058-ddad4a821aee49586fd213ab9a4a77bb2826a076
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 2cb6b54085 fix(proto): preserve CSV sink writer options (#25058)
This branch includes the following new commits:
new 2cb6b54085 fix(proto): preserve CSV sink writer options (#25058)
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(proto): preserve CSV sink writer options (#25058)
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-25058-ddad4a821aee49586fd213ab9a4a77bb2826a076
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 2cb6b54085ad36f412717b6a79b683d78ae2f48e
Author: Burak Şen <bu...@gmail.com>
AuthorDate: Wed Sep 16 13:50:00 2026 +0000
fix(proto): preserve CSV sink writer options (#25058)
## Which issue does this PR close?
- Closes #24623.
## Rationale for this change
The CSV sink protobuf hooks accessed state indirectly, allowing new
fields to be omitted silently. Auditing the writer options found that
compression level, timezone-aware timestamp format, and line terminator
were not represented on the wire, so those settings changed after a
physical-plan round trip.
## What changes are included in this PR?
- Exhaustively destructure `CsvSink`, `CsvSinkExecNode`, and the nested
CSV sink/writer option messages.
- Add protobuf fields for compression level, timezone-aware timestamp
format, and line terminator.
- Make all five optional date/time formats presence-aware so `None` and
`Some("")` remain distinct.
- Add a DataFusion 56.0 upgrade guide for the generated
`CsvWriterOptions` API changes.
- Regenerate the protobuf models.
The protobuf wire format remains backward compatible. The generated Rust
API changes are documented in the upgrade guide.
## What is the testing strategy for this PR?
Expanded the CSV sink round-trip test to cover every writer option,
optional defaults, explicitly empty formats, and malformed line
terminators.
Validated with:
- `cargo test -p datafusion-proto --test proto_integration
roundtrip_csv_sink`
- `cargo clippy -p datafusion-datasource-csv -p datafusion-proto-common
-p datafusion-proto --all-features --tests -- -D warnings`
- `cargo fmt --all --check`
## Are there any user-facing changes?
CSV sinks now retain compression level, timezone-aware timestamp
formatting, line terminators, and explicitly empty date/time formats
across protobuf plan round trips. Invalid wire terminators return an
error.
The generated public `CsvWriterOptions` gains three fields, and its
existing date/time format fields change from `String` to
`Option<String>`. Migration is documented in the DataFusion 56.0 upgrade
guide. The protobuf wire format remains backward compatible.
---------
Co-authored-by: Kumar Ujjawal <uj...@gmail.com>
---
datafusion/datasource-csv/src/file_format.rs | 52 +++++--
.../proto-common/proto/datafusion_common.proto | 14 +-
datafusion/proto-common/src/from_proto/mod.rs | 171 +++++++++++++--------
datafusion/proto-common/src/generated/pbjson.rs | 99 +++++++++---
datafusion/proto-common/src/generated/prost.rs | 25 ++-
datafusion/proto-common/src/to_proto/mod.rs | 62 ++++----
.../src/generated/datafusion_proto_common.rs | 25 ++-
datafusion/proto/tests/cases/plans/sinks.rs | 111 ++++++++++++-
docs/source/library-user-guide/upgrading/56.0.0.md | 37 +++++
9 files changed, 439 insertions(+), 157 deletions(-)
diff --git a/datafusion/datasource-csv/src/file_format.rs b/datafusion/datasource-csv/src/file_format.rs
index 1f7da7e849..15e24df4bf 100644
--- a/datafusion/datasource-csv/src/file_format.rs
+++ b/datafusion/datasource-csv/src/file_format.rs
@@ -861,6 +861,13 @@ impl DataSink for CsvSink {
use datafusion_proto_models::protobuf;
use protobuf::physical_plan_node::PhysicalPlanType;
+ // Keep an exhaustive guard in the active hook while centralizing the
+ // field mapping in the exhaustive `TryFrom<&CsvSink>` below.
+ let Self {
+ config: _,
+ writer_options: _,
+ } = self;
+
let input = ctx.encode_child(exec.input())?;
let sort_order = exec.encode_sort_order(ctx)?;
let sink = protobuf::CsvSink::try_from(self)?;
@@ -881,9 +888,13 @@ impl TryFrom<&CsvSink> for datafusion_proto_models::protobuf::CsvSink {
type Error = DataFusionError;
fn try_from(value: &CsvSink) -> Result<Self> {
+ let CsvSink {
+ config,
+ writer_options,
+ } = value;
Ok(Self {
- config: Some(value.config().try_into()?),
- writer_options: Some(value.writer_options().try_into()?),
+ config: Some(config.try_into()?),
+ writer_options: Some(writer_options.try_into()?),
})
}
}
@@ -893,14 +904,16 @@ impl TryFrom<&datafusion_proto_models::protobuf::CsvSink> for CsvSink {
type Error = DataFusionError;
fn try_from(value: &datafusion_proto_models::protobuf::CsvSink) -> Result<Self> {
- let config =
- FileSinkConfig::try_from(value.config.as_ref().ok_or_else(|| {
- datafusion_common::internal_datafusion_err!(
- "CsvSink is missing required field 'config'"
- )
- })?)?;
- let writer_options = value
- .writer_options
+ let datafusion_proto_models::protobuf::CsvSink {
+ config,
+ writer_options,
+ } = value;
+ let config = FileSinkConfig::try_from(config.as_ref().ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!(
+ "CsvSink is missing required field 'config'"
+ )
+ })?)?;
+ let writer_options = writer_options
.as_ref()
.ok_or_else(|| {
datafusion_common::internal_datafusion_err!(
@@ -927,19 +940,24 @@ impl CsvSink {
protobuf::physical_plan_node::PhysicalPlanType::CsvSink,
"CsvSink",
);
- let input = ctx.decode_required_child(
- sink_node.input.as_deref(),
- "CsvSinkExecNode",
- "input",
- )?;
- let proto_sink = sink_node.sink.as_ref().ok_or_else(|| {
+ let protobuf::CsvSinkExecNode {
+ input,
+ sink,
+ // The output schema is recomputed by `DataSinkExec::new`.
+ sink_schema: _,
+ sort_order,
+ } = sink_node.as_ref();
+
+ let input =
+ ctx.decode_required_child(input.as_deref(), "CsvSinkExecNode", "input")?;
+ let proto_sink = sink.as_ref().ok_or_else(|| {
datafusion_common::internal_datafusion_err!(
"CsvSinkExecNode is missing required field 'sink'"
)
})?;
let data_sink = CsvSink::try_from(proto_sink)?;
let sort_order = DataSinkExec::decode_sort_order(
- sink_node.sort_order.as_ref(),
+ sort_order.as_ref(),
ctx,
input.schema().as_ref(),
)?;
diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto
index 712212f6b6..bdabb0777e 100644
--- a/datafusion/proto-common/proto/datafusion_common.proto
+++ b/datafusion/proto-common/proto/datafusion_common.proto
@@ -450,13 +450,13 @@ message CsvWriterOptions {
// Whether to write column names as file headers. Defaults to `true`
bool has_header = 3;
// Optional date format for date arrays
- string date_format = 4;
+ optional string date_format = 4;
// Optional datetime format for datetime arrays
- string datetime_format = 5;
+ optional string datetime_format = 5;
// Optional timestamp format for timestamp arrays
- string timestamp_format = 6;
+ optional string timestamp_format = 6;
// Optional time format for time arrays
- string time_format = 7;
+ optional string time_format = 7;
// Optional value to represent null
string null_value = 8;
// Optional quote. Defaults to `b'"'`
@@ -471,6 +471,12 @@ message CsvWriterOptions {
bool ignore_leading_whitespace = 13;
// Whether to ignore trailing whitespace in string values
bool ignore_trailing_whitespace = 14;
+ // Optional compression level
+ optional uint32 compression_level = 15;
+ // Optional timestamp format for timestamp with timezone arrays
+ optional string timestamp_tz_format = 16;
+ // Optional line terminator. Empty defaults to LF; valid values are one byte or CRLF
+ bytes terminator = 17;
}
// Options controlling CSV format
diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs
index 92506ad92b..8ddb1cf967 100644
--- a/datafusion/proto-common/src/from_proto/mod.rs
+++ b/datafusion/proto-common/src/from_proto/mod.rs
@@ -22,6 +22,7 @@ use crate::common::proto_error;
use crate::protobuf_common as protobuf;
use arrow::array::{ArrayRef, AsArray};
use arrow::buffer::Buffer;
+use arrow::csv::writer::Terminator;
use arrow::csv::{QuoteStyle, WriterBuilder};
use arrow::datatypes::{
DataType, Field, IntervalDayTimeType, IntervalMonthDayNanoType, IntervalUnit, Schema,
@@ -985,9 +986,113 @@ impl TryFrom<&protobuf::CsvWriterOptions> for CsvWriterOptions {
fn try_from(
opts: &protobuf::CsvWriterOptions,
) -> datafusion_common::Result<Self, Self::Error> {
- let write_options = csv_writer_options_from_proto(opts)?;
- let compression: CompressionTypeVariant = opts.compression().into();
- Ok(CsvWriterOptions::new(write_options, compression))
+ let protobuf::CsvWriterOptions {
+ compression,
+ delimiter,
+ has_header,
+ date_format,
+ datetime_format,
+ timestamp_format,
+ time_format,
+ null_value,
+ quote,
+ escape,
+ double_quote,
+ quote_style,
+ ignore_leading_whitespace,
+ ignore_trailing_whitespace,
+ compression_level,
+ timestamp_tz_format,
+ terminator,
+ } = opts;
+
+ let mut writer_options = WriterBuilder::new();
+ if !delimiter.is_empty() {
+ if let Some(delimiter) = delimiter.chars().next() {
+ if delimiter.is_ascii() {
+ writer_options = writer_options.with_delimiter(delimiter as u8);
+ } else {
+ return Err(proto_error("CSV Delimiter is not ASCII"));
+ }
+ } else {
+ return Err(proto_error("Error parsing CSV Delimiter"));
+ }
+ }
+ if !quote.is_empty() {
+ if let Some(quote) = quote.chars().next() {
+ if quote.is_ascii() {
+ writer_options = writer_options.with_quote(quote as u8);
+ } else {
+ return Err(proto_error("CSV Quote is not ASCII"));
+ }
+ } else {
+ return Err(proto_error("Error parsing CSV Quote"));
+ }
+ }
+ if !escape.is_empty() {
+ if let Some(escape) = escape.chars().next() {
+ if escape.is_ascii() {
+ writer_options = writer_options.with_escape(escape as u8);
+ } else {
+ return Err(proto_error("CSV Escape is not ASCII"));
+ }
+ } else {
+ return Err(proto_error("Error parsing CSV Escape"));
+ }
+ }
+ let quote_style = match protobuf::CsvQuoteStyle::try_from(*quote_style) {
+ Ok(protobuf::CsvQuoteStyle::Always) => QuoteStyle::Always,
+ Ok(protobuf::CsvQuoteStyle::NonNumeric) => QuoteStyle::NonNumeric,
+ Ok(protobuf::CsvQuoteStyle::Never) => QuoteStyle::Never,
+ Ok(protobuf::CsvQuoteStyle::Necessary) => QuoteStyle::Necessary,
+ _ => {
+ return Err(proto_error(
+ "Unknown quote style, must be one of: 'Always', 'NonNumeric', 'Never', 'Necessary'",
+ ));
+ }
+ };
+ writer_options = writer_options
+ .with_header(*has_header)
+ .with_null(null_value.clone())
+ .with_double_quote(*double_quote)
+ .with_quote_style(quote_style)
+ .with_ignore_leading_whitespace(*ignore_leading_whitespace)
+ .with_ignore_trailing_whitespace(*ignore_trailing_whitespace);
+ // Presence-aware: `Some("")` is a valid format distinct from `None`.
+ if let Some(date_format) = date_format {
+ writer_options = writer_options.with_date_format(date_format.clone());
+ }
+ if let Some(datetime_format) = datetime_format {
+ writer_options = writer_options.with_datetime_format(datetime_format.clone());
+ }
+ if let Some(timestamp_format) = timestamp_format {
+ writer_options =
+ writer_options.with_timestamp_format(timestamp_format.clone());
+ }
+ if let Some(timestamp_tz_format) = timestamp_tz_format {
+ writer_options =
+ writer_options.with_timestamp_tz_format(timestamp_tz_format.clone());
+ }
+ if let Some(time_format) = time_format {
+ writer_options = writer_options.with_time_format(time_format.clone());
+ }
+ writer_options = match terminator.as_slice() {
+ [] => writer_options,
+ [byte] => writer_options.with_line_terminator(Terminator::Any(*byte)),
+ [b'\r', b'\n'] => writer_options.with_line_terminator(Terminator::CRLF),
+ _ => {
+ return Err(proto_error("CSV line terminator must be one byte or CRLF"));
+ }
+ };
+
+ let compression = protobuf::CompressionTypeVariant::try_from(*compression)
+ .unwrap_or_default()
+ .into();
+ Ok(CsvWriterOptions {
+ writer_options,
+ compression,
+ compression_level: *compression_level,
+ })
}
}
@@ -1324,66 +1429,6 @@ where
.collect::<datafusion_common::Result<_, _>>()
}
-pub(crate) fn csv_writer_options_from_proto(
- writer_options: &protobuf::CsvWriterOptions,
-) -> datafusion_common::Result<WriterBuilder> {
- let mut builder = WriterBuilder::new();
- if !writer_options.delimiter.is_empty() {
- if let Some(delimiter) = writer_options.delimiter.chars().next() {
- if delimiter.is_ascii() {
- builder = builder.with_delimiter(delimiter as u8);
- } else {
- return Err(proto_error("CSV Delimiter is not ASCII"));
- }
- } else {
- return Err(proto_error("Error parsing CSV Delimiter"));
- }
- }
- if !writer_options.quote.is_empty() {
- if let Some(quote) = writer_options.quote.chars().next() {
- if quote.is_ascii() {
- builder = builder.with_quote(quote as u8);
- } else {
- return Err(proto_error("CSV Quote is not ASCII"));
- }
- } else {
- return Err(proto_error("Error parsing CSV Quote"));
- }
- }
- if !writer_options.escape.is_empty() {
- if let Some(escape) = writer_options.escape.chars().next() {
- if escape.is_ascii() {
- builder = builder.with_escape(escape as u8);
- } else {
- return Err(proto_error("CSV Escape is not ASCII"));
- }
- } else {
- return Err(proto_error("Error parsing CSV Escape"));
- }
- }
- let quote_style = match protobuf::CsvQuoteStyle::try_from(writer_options.quote_style)
- {
- Ok(protobuf::CsvQuoteStyle::Always) => QuoteStyle::Always,
- Ok(protobuf::CsvQuoteStyle::NonNumeric) => QuoteStyle::NonNumeric,
- Ok(protobuf::CsvQuoteStyle::Never) => QuoteStyle::Never,
- Ok(protobuf::CsvQuoteStyle::Necessary) => QuoteStyle::Necessary,
- _ => Err(proto_error(
- "Unknown quote style, must be one of: 'Always', 'NonNumeric', 'Never', 'Necessary'",
- ))?,
- };
- Ok(builder
- .with_header(writer_options.has_header)
- .with_date_format(writer_options.date_format.clone())
- .with_datetime_format(writer_options.datetime_format.clone())
- .with_timestamp_format(writer_options.timestamp_format.clone())
- .with_time_format(writer_options.time_format.clone())
- .with_null(writer_options.null_value.clone())
- .with_double_quote(writer_options.double_quote)
- .with_quote_style(quote_style)
- .with_ignore_leading_whitespace(writer_options.ignore_leading_whitespace)
- .with_ignore_trailing_whitespace(writer_options.ignore_trailing_whitespace))
-}
-
#[cfg(test)]
mod tests {
use datafusion_common::config::{
diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs
index 2e94368bfd..a1d86c5fe3 100644
--- a/datafusion/proto-common/src/generated/pbjson.rs
+++ b/datafusion/proto-common/src/generated/pbjson.rs
@@ -2276,16 +2276,16 @@ impl serde::Serialize for CsvWriterOptions {
if self.has_header {
len += 1;
}
- if !self.date_format.is_empty() {
+ if self.date_format.is_some() {
len += 1;
}
- if !self.datetime_format.is_empty() {
+ if self.datetime_format.is_some() {
len += 1;
}
- if !self.timestamp_format.is_empty() {
+ if self.timestamp_format.is_some() {
len += 1;
}
- if !self.time_format.is_empty() {
+ if self.time_format.is_some() {
len += 1;
}
if !self.null_value.is_empty() {
@@ -2309,6 +2309,15 @@ impl serde::Serialize for CsvWriterOptions {
if self.ignore_trailing_whitespace {
len += 1;
}
+ if self.compression_level.is_some() {
+ len += 1;
+ }
+ if self.timestamp_tz_format.is_some() {
+ len += 1;
+ }
+ if !self.terminator.is_empty() {
+ len += 1;
+ }
let mut struct_ser = serializer.serialize_struct("datafusion_common.CsvWriterOptions", len)?;
if self.compression != 0 {
let v = CompressionTypeVariant::try_from(self.compression)
@@ -2321,17 +2330,17 @@ impl serde::Serialize for CsvWriterOptions {
if self.has_header {
struct_ser.serialize_field("hasHeader", &self.has_header)?;
}
- if !self.date_format.is_empty() {
- struct_ser.serialize_field("dateFormat", &self.date_format)?;
+ if let Some(v) = self.date_format.as_ref() {
+ struct_ser.serialize_field("dateFormat", v)?;
}
- if !self.datetime_format.is_empty() {
- struct_ser.serialize_field("datetimeFormat", &self.datetime_format)?;
+ if let Some(v) = self.datetime_format.as_ref() {
+ struct_ser.serialize_field("datetimeFormat", v)?;
}
- if !self.timestamp_format.is_empty() {
- struct_ser.serialize_field("timestampFormat", &self.timestamp_format)?;
+ if let Some(v) = self.timestamp_format.as_ref() {
+ struct_ser.serialize_field("timestampFormat", v)?;
}
- if !self.time_format.is_empty() {
- struct_ser.serialize_field("timeFormat", &self.time_format)?;
+ if let Some(v) = self.time_format.as_ref() {
+ struct_ser.serialize_field("timeFormat", v)?;
}
if !self.null_value.is_empty() {
struct_ser.serialize_field("nullValue", &self.null_value)?;
@@ -2356,6 +2365,17 @@ impl serde::Serialize for CsvWriterOptions {
if self.ignore_trailing_whitespace {
struct_ser.serialize_field("ignoreTrailingWhitespace", &self.ignore_trailing_whitespace)?;
}
+ if let Some(v) = self.compression_level.as_ref() {
+ struct_ser.serialize_field("compressionLevel", v)?;
+ }
+ if let Some(v) = self.timestamp_tz_format.as_ref() {
+ struct_ser.serialize_field("timestampTzFormat", v)?;
+ }
+ if !self.terminator.is_empty() {
+ #[allow(clippy::needless_borrow)]
+ #[allow(clippy::needless_borrows_for_generic_args)]
+ struct_ser.serialize_field("terminator", pbjson::private::base64::encode(&self.terminator).as_str())?;
+ }
struct_ser.end()
}
}
@@ -2390,6 +2410,11 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
"ignoreLeadingWhitespace",
"ignore_trailing_whitespace",
"ignoreTrailingWhitespace",
+ "compression_level",
+ "compressionLevel",
+ "timestamp_tz_format",
+ "timestampTzFormat",
+ "terminator",
];
#[allow(clippy::enum_variant_names)]
@@ -2408,6 +2433,9 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
QuoteStyle,
IgnoreLeadingWhitespace,
IgnoreTrailingWhitespace,
+ CompressionLevel,
+ TimestampTzFormat,
+ Terminator,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -2443,6 +2471,9 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
"quoteStyle" | "quote_style" => Ok(GeneratedField::QuoteStyle),
"ignoreLeadingWhitespace" | "ignore_leading_whitespace" => Ok(GeneratedField::IgnoreLeadingWhitespace),
"ignoreTrailingWhitespace" | "ignore_trailing_whitespace" => Ok(GeneratedField::IgnoreTrailingWhitespace),
+ "compressionLevel" | "compression_level" => Ok(GeneratedField::CompressionLevel),
+ "timestampTzFormat" | "timestamp_tz_format" => Ok(GeneratedField::TimestampTzFormat),
+ "terminator" => Ok(GeneratedField::Terminator),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
}
}
@@ -2476,6 +2507,9 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
let mut quote_style__ = None;
let mut ignore_leading_whitespace__ = None;
let mut ignore_trailing_whitespace__ = None;
+ let mut compression_level__ = None;
+ let mut timestamp_tz_format__ = None;
+ let mut terminator__ = None;
while let Some(k) = map_.next_key()? {
match k {
GeneratedField::Compression => {
@@ -2500,25 +2534,25 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
if date_format__.is_some() {
return Err(serde::de::Error::duplicate_field("dateFormat"));
}
- date_format__ = Some(map_.next_value()?);
+ date_format__ = map_.next_value()?;
}
GeneratedField::DatetimeFormat => {
if datetime_format__.is_some() {
return Err(serde::de::Error::duplicate_field("datetimeFormat"));
}
- datetime_format__ = Some(map_.next_value()?);
+ datetime_format__ = map_.next_value()?;
}
GeneratedField::TimestampFormat => {
if timestamp_format__.is_some() {
return Err(serde::de::Error::duplicate_field("timestampFormat"));
}
- timestamp_format__ = Some(map_.next_value()?);
+ timestamp_format__ = map_.next_value()?;
}
GeneratedField::TimeFormat => {
if time_format__.is_some() {
return Err(serde::de::Error::duplicate_field("timeFormat"));
}
- time_format__ = Some(map_.next_value()?);
+ time_format__ = map_.next_value()?;
}
GeneratedField::NullValue => {
if null_value__.is_some() {
@@ -2562,16 +2596,38 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
}
ignore_trailing_whitespace__ = Some(map_.next_value()?);
}
+ GeneratedField::CompressionLevel => {
+ if compression_level__.is_some() {
+ return Err(serde::de::Error::duplicate_field("compressionLevel"));
+ }
+ compression_level__ =
+ map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0)
+ ;
+ }
+ GeneratedField::TimestampTzFormat => {
+ if timestamp_tz_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("timestampTzFormat"));
+ }
+ timestamp_tz_format__ = map_.next_value()?;
+ }
+ GeneratedField::Terminator => {
+ if terminator__.is_some() {
+ return Err(serde::de::Error::duplicate_field("terminator"));
+ }
+ terminator__ =
+ Some(map_.next_value::<::pbjson::private::BytesDeserialize<_>>()?.0)
+ ;
+ }
}
}
Ok(CsvWriterOptions {
compression: compression__.unwrap_or_default(),
delimiter: delimiter__.unwrap_or_default(),
has_header: has_header__.unwrap_or_default(),
- date_format: date_format__.unwrap_or_default(),
- datetime_format: datetime_format__.unwrap_or_default(),
- timestamp_format: timestamp_format__.unwrap_or_default(),
- time_format: time_format__.unwrap_or_default(),
+ date_format: date_format__,
+ datetime_format: datetime_format__,
+ timestamp_format: timestamp_format__,
+ time_format: time_format__,
null_value: null_value__.unwrap_or_default(),
quote: quote__.unwrap_or_default(),
escape: escape__.unwrap_or_default(),
@@ -2579,6 +2635,9 @@ impl<'de> serde::Deserialize<'de> for CsvWriterOptions {
quote_style: quote_style__.unwrap_or_default(),
ignore_leading_whitespace: ignore_leading_whitespace__.unwrap_or_default(),
ignore_trailing_whitespace: ignore_trailing_whitespace__.unwrap_or_default(),
+ compression_level: compression_level__,
+ timestamp_tz_format: timestamp_tz_format__,
+ terminator: terminator__.unwrap_or_default(),
})
}
}
diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs
index c0e79aec6d..f856c1aa39 100644
--- a/datafusion/proto-common/src/generated/prost.rs
+++ b/datafusion/proto-common/src/generated/prost.rs
@@ -598,17 +598,17 @@ pub struct CsvWriterOptions {
#[prost(bool, tag = "3")]
pub has_header: bool,
/// Optional date format for date arrays
- #[prost(string, tag = "4")]
- pub date_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "4")]
+ pub date_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional datetime format for datetime arrays
- #[prost(string, tag = "5")]
- pub datetime_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "5")]
+ pub datetime_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional timestamp format for timestamp arrays
- #[prost(string, tag = "6")]
- pub timestamp_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "6")]
+ pub timestamp_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional time format for time arrays
- #[prost(string, tag = "7")]
- pub time_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "7")]
+ pub time_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional value to represent null
#[prost(string, tag = "8")]
pub null_value: ::prost::alloc::string::String,
@@ -630,6 +630,15 @@ pub struct CsvWriterOptions {
/// Whether to ignore trailing whitespace in string values
#[prost(bool, tag = "14")]
pub ignore_trailing_whitespace: bool,
+ /// Optional compression level
+ #[prost(uint32, optional, tag = "15")]
+ pub compression_level: ::core::option::Option<u32>,
+ /// Optional timestamp format for timestamp with timezone arrays
+ #[prost(string, optional, tag = "16")]
+ pub timestamp_tz_format: ::core::option::Option<::prost::alloc::string::String>,
+ /// Optional line terminator. Empty defaults to LF; valid values are one byte or CRLF
+ #[prost(bytes = "vec", tag = "17")]
+ pub terminator: ::prost::alloc::vec::Vec<u8>,
}
/// Options controlling CSV format
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs
index 7f95e03f41..2d47a84709 100644
--- a/datafusion/proto-common/src/to_proto/mod.rs
+++ b/datafusion/proto-common/src/to_proto/mod.rs
@@ -23,7 +23,8 @@ use crate::protobuf_common::{
EmptyMessage, arrow_type::ArrowTypeEnum, scalar_value::Value,
};
use arrow::array::{ArrayRef, RecordBatch};
-use arrow::csv::{QuoteStyle, WriterBuilder};
+use arrow::csv::QuoteStyle;
+use arrow::csv::writer::Terminator;
use arrow::datatypes::{
DataType, Field, IntervalDayTimeType, IntervalMonthDayNanoType, IntervalUnit, Schema,
SchemaRef, TimeUnit, UnionMode,
@@ -872,10 +873,37 @@ impl TryFrom<&CsvWriterOptions> for protobuf::CsvWriterOptions {
type Error = DataFusionError;
fn try_from(opts: &CsvWriterOptions) -> datafusion_common::Result<Self, Self::Error> {
- Ok(csv_writer_options_to_proto(
- &opts.writer_options,
- &opts.compression,
- ))
+ let CsvWriterOptions {
+ writer_options,
+ compression,
+ compression_level,
+ } = opts;
+ let compression: protobuf::CompressionTypeVariant = compression.into();
+ let quote_style: protobuf::CsvQuoteStyle = writer_options.quote_style().into();
+ let terminator = match writer_options.line_terminator() {
+ Terminator::CRLF => b"\r\n".to_vec(),
+ Terminator::Any(byte) => vec![*byte],
+ };
+
+ Ok(protobuf::CsvWriterOptions {
+ compression: compression.into(),
+ delimiter: (writer_options.delimiter() as char).to_string(),
+ has_header: writer_options.header(),
+ date_format: writer_options.date_format().map(str::to_owned),
+ datetime_format: writer_options.datetime_format().map(str::to_owned),
+ timestamp_format: writer_options.timestamp_format().map(str::to_owned),
+ time_format: writer_options.time_format().map(str::to_owned),
+ null_value: writer_options.null().to_owned(),
+ quote: (writer_options.quote() as char).to_string(),
+ escape: (writer_options.escape() as char).to_string(),
+ double_quote: writer_options.double_quote(),
+ quote_style: quote_style.into(),
+ ignore_leading_whitespace: writer_options.ignore_leading_whitespace(),
+ ignore_trailing_whitespace: writer_options.ignore_trailing_whitespace(),
+ compression_level: *compression_level,
+ timestamp_tz_format: writer_options.timestamp_tz_format().map(str::to_owned),
+ terminator,
+ })
}
}
@@ -1175,27 +1203,3 @@ where
.map(|field| field.as_ref().try_into())
.collect::<Result<Vec<_>, Error>>()
}
-
-pub(crate) fn csv_writer_options_to_proto(
- csv_options: &WriterBuilder,
- compression: &CompressionTypeVariant,
-) -> protobuf::CsvWriterOptions {
- let compression: protobuf::CompressionTypeVariant = compression.into();
- let quote_style: protobuf::CsvQuoteStyle = csv_options.quote_style().into();
- protobuf::CsvWriterOptions {
- compression: compression.into(),
- delimiter: (csv_options.delimiter() as char).to_string(),
- has_header: csv_options.header(),
- date_format: csv_options.date_format().unwrap_or("").to_owned(),
- datetime_format: csv_options.datetime_format().unwrap_or("").to_owned(),
- timestamp_format: csv_options.timestamp_format().unwrap_or("").to_owned(),
- time_format: csv_options.time_format().unwrap_or("").to_owned(),
- null_value: csv_options.null().to_owned(),
- quote: (csv_options.quote() as char).to_string(),
- escape: (csv_options.escape() as char).to_string(),
- double_quote: csv_options.double_quote(),
- quote_style: quote_style.into(),
- ignore_leading_whitespace: csv_options.ignore_leading_whitespace(),
- ignore_trailing_whitespace: csv_options.ignore_trailing_whitespace(),
- }
-}
diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs
index c0e79aec6d..f856c1aa39 100644
--- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs
+++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs
@@ -598,17 +598,17 @@ pub struct CsvWriterOptions {
#[prost(bool, tag = "3")]
pub has_header: bool,
/// Optional date format for date arrays
- #[prost(string, tag = "4")]
- pub date_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "4")]
+ pub date_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional datetime format for datetime arrays
- #[prost(string, tag = "5")]
- pub datetime_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "5")]
+ pub datetime_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional timestamp format for timestamp arrays
- #[prost(string, tag = "6")]
- pub timestamp_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "6")]
+ pub timestamp_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional time format for time arrays
- #[prost(string, tag = "7")]
- pub time_format: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "7")]
+ pub time_format: ::core::option::Option<::prost::alloc::string::String>,
/// Optional value to represent null
#[prost(string, tag = "8")]
pub null_value: ::prost::alloc::string::String,
@@ -630,6 +630,15 @@ pub struct CsvWriterOptions {
/// Whether to ignore trailing whitespace in string values
#[prost(bool, tag = "14")]
pub ignore_trailing_whitespace: bool,
+ /// Optional compression level
+ #[prost(uint32, optional, tag = "15")]
+ pub compression_level: ::core::option::Option<u32>,
+ /// Optional timestamp format for timestamp with timezone arrays
+ #[prost(string, optional, tag = "16")]
+ pub timestamp_tz_format: ::core::option::Option<::prost::alloc::string::String>,
+ /// Optional line terminator. Empty defaults to LF; valid values are one byte or CRLF
+ #[prost(bytes = "vec", tag = "17")]
+ pub terminator: ::prost::alloc::vec::Vec<u8>,
}
/// Options controlling CSV format
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
diff --git a/datafusion/proto/tests/cases/plans/sinks.rs b/datafusion/proto/tests/cases/plans/sinks.rs
index 63bf11dd30..8d8b736000 100644
--- a/datafusion/proto/tests/cases/plans/sinks.rs
+++ b/datafusion/proto/tests/cases/plans/sinks.rs
@@ -18,7 +18,8 @@
//! Data sinks and their file sink configurations.
use super::{roundtrip_test, roundtrip_test_and_return};
-use arrow::csv::WriterBuilder;
+use arrow::csv::writer::Terminator;
+use arrow::csv::{QuoteStyle, WriterBuilder};
use async_trait::async_trait;
use datafusion::arrow::compute::kernels::sort::SortOptions;
use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
@@ -51,6 +52,7 @@ use datafusion_proto::physical_plan::{
};
use datafusion_proto::protobuf;
use datafusion_proto::protobuf::PhysicalPlanNode;
+use datafusion_proto_common::protobuf_common::CsvWriterOptions as ProtoCsvWriterOptions;
use std::fmt::Formatter;
use std::sync::Arc;
use std::vec;
@@ -291,11 +293,27 @@ fn roundtrip_csv_sink() -> Result<()> {
file_extension: "csv".into(),
file_output_mode: FileOutputMode::Directory,
};
+ let writer_options = WriterBuilder::default()
+ .with_delimiter(b'|')
+ .with_header(false)
+ .with_quote(b'\'')
+ .with_escape(b'!')
+ .with_double_quote(false)
+ .with_date_format("%Y/%m/%d".into())
+ .with_datetime_format("%Y/%m/%d %H:%M:%S".into())
+ .with_timestamp_format("%s".into())
+ .with_timestamp_tz_format("%Y-%m-%dT%H:%M:%S%:z".into())
+ .with_time_format("%H-%M-%S".into())
+ .with_null("NULL".into())
+ .with_quote_style(QuoteStyle::Always)
+ .with_ignore_leading_whitespace(true)
+ .with_ignore_trailing_whitespace(true)
+ .with_line_terminator(Terminator::CRLF);
let data_sink = Arc::new(CsvSink::new(
file_sink_config,
- CsvWriterOptions::new(WriterBuilder::default(), CompressionTypeVariant::ZSTD),
+ CsvWriterOptions::new_with_level(writer_options, CompressionTypeVariant::ZSTD, 7),
));
- let sort_order = [PhysicalSortRequirement::new(
+ let sort_order: LexRequirement = [PhysicalSortRequirement::new(
Arc::new(Column::new("plan_type", 0)),
Some(SortOptions {
descending: true,
@@ -309,18 +327,95 @@ fn roundtrip_csv_sink() -> Result<()> {
let proto_converter = DefaultPhysicalProtoConverter {};
let roundtrip_plan = roundtrip_test_and_return(
- Arc::new(DataSinkExec::new(input, data_sink, Some(sort_order))),
+ Arc::new(DataSinkExec::new(
+ input,
+ data_sink,
+ Some(sort_order.clone()),
+ )),
&ctx,
&codec,
&proto_converter,
)?;
- let roundtrip_plan = roundtrip_plan.downcast_ref::<DataSinkExec>().unwrap();
- let csv_sink = roundtrip_plan.sink().downcast_ref::<CsvSink>().unwrap();
+ let roundtrip_plan =
+ roundtrip_plan
+ .downcast_ref::<DataSinkExec>()
+ .ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!("Expected DataSinkExec")
+ })?;
+ let csv_sink = roundtrip_plan
+ .sink()
+ .downcast_ref::<CsvSink>()
+ .ok_or_else(|| datafusion_common::internal_datafusion_err!("Expected CsvSink"))?;
+ assert_eq!(csv_sink.config().insert_op, InsertOp::Overwrite);
+ assert!(csv_sink.config().keep_partition_by_columns);
assert_eq!(
- CompressionTypeVariant::ZSTD,
- csv_sink.writer_options().compression
+ csv_sink.config().file_output_mode,
+ FileOutputMode::Directory
+ );
+
+ let options = csv_sink.writer_options();
+ assert_eq!(options.compression, CompressionTypeVariant::ZSTD);
+ assert_eq!(options.compression_level, Some(7));
+ let writer = &options.writer_options;
+ assert_eq!(writer.delimiter(), b'|');
+ assert!(!writer.header());
+ assert_eq!(writer.quote(), b'\'');
+ assert_eq!(writer.escape(), b'!');
+ assert!(!writer.double_quote());
+ assert_eq!(writer.date_format(), Some("%Y/%m/%d"));
+ assert_eq!(writer.datetime_format(), Some("%Y/%m/%d %H:%M:%S"));
+ assert_eq!(writer.timestamp_format(), Some("%s"));
+ assert_eq!(writer.timestamp_tz_format(), Some("%Y-%m-%dT%H:%M:%S%:z"));
+ assert_eq!(writer.time_format(), Some("%H-%M-%S"));
+ assert_eq!(writer.null(), "NULL");
+ assert!(matches!(writer.quote_style(), QuoteStyle::Always));
+ assert!(writer.ignore_leading_whitespace());
+ assert!(writer.ignore_trailing_whitespace());
+ assert!(matches!(writer.line_terminator(), Terminator::CRLF));
+ assert_eq!(roundtrip_plan.sort_order(), &Some(sort_order));
+
+ let unset = CsvWriterOptions::new(
+ WriterBuilder::default(),
+ CompressionTypeVariant::UNCOMPRESSED,
+ );
+ let unset = CsvWriterOptions::try_from(&ProtoCsvWriterOptions::try_from(&unset)?)?;
+ assert_eq!(unset.writer_options.date_format(), None);
+ assert_eq!(unset.writer_options.datetime_format(), None);
+ assert_eq!(unset.writer_options.timestamp_format(), None);
+ assert_eq!(unset.writer_options.timestamp_tz_format(), None);
+ assert_eq!(unset.writer_options.time_format(), None);
+
+ // An explicit empty format is distinct from an unset one: Arrow writes an
+ // empty value for it, so it must not collapse to `None` on the wire.
+ let empty = CsvWriterOptions::new(
+ WriterBuilder::default()
+ .with_date_format(String::new())
+ .with_datetime_format(String::new())
+ .with_timestamp_format(String::new())
+ .with_timestamp_tz_format(String::new())
+ .with_time_format(String::new()),
+ CompressionTypeVariant::UNCOMPRESSED,
);
+ let empty = CsvWriterOptions::try_from(&ProtoCsvWriterOptions::try_from(&empty)?)?;
+ assert_eq!(empty.writer_options.date_format(), Some(""));
+ assert_eq!(empty.writer_options.datetime_format(), Some(""));
+ assert_eq!(empty.writer_options.timestamp_format(), Some(""));
+ assert_eq!(empty.writer_options.timestamp_tz_format(), Some(""));
+ assert_eq!(empty.writer_options.time_format(), Some(""));
+
+ let defaults = CsvWriterOptions::try_from(&ProtoCsvWriterOptions::default())?;
+ assert_eq!(defaults.compression_level, None);
+ assert!(matches!(
+ defaults.writer_options.line_terminator(),
+ Terminator::Any(b'\n')
+ ));
+
+ let malformed = ProtoCsvWriterOptions {
+ terminator: b"\r\r".to_vec(),
+ ..Default::default()
+ };
+ assert!(CsvWriterOptions::try_from(&malformed).is_err());
Ok(())
}
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index ef51e4c9a7..20cd0809ee 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -227,6 +227,43 @@ The protobuf wire format remains backward compatible.
See [PR #24945](https://github.com/apache/datafusion/pull/24945) for details.
+### Generated protobuf `CsvWriterOptions` changed
+
+The generated `CsvWriterOptions` now includes `compression_level`,
+`timestamp_tz_format`, and `terminator`. Its four existing format fields changed
+from `String` to `Option<String>` to distinguish unset from explicitly empty
+formats. This affects exhaustive struct literals and direct format-field access.
+
+Wrap existing formats in `Some` or use `None`, and initialize the new fields
+(explicitly or through `Default`):
+
+```rust,ignore
+// Before (unchanged fields omitted)
+CsvWriterOptions {
+ date_format: "%Y-%m-%d".to_string(),
+ datetime_format: String::new(),
+ timestamp_format: String::new(),
+ time_format: String::new(),
+ // ...
+}
+
+// After
+CsvWriterOptions {
+ date_format: Some("%Y-%m-%d".to_string()),
+ datetime_format: None,
+ timestamp_format: None,
+ time_format: None,
+ compression_level: None,
+ timestamp_tz_format: None,
+ terminator: Vec::new(),
+ // ...
+}
+```
+
+The protobuf wire format remains backward compatible.
+
+See [PR #25058](https://github.com/apache/datafusion/pull/25058) for details.
+
### `DefaultStatisticsProvider` is deprecated
`datafusion_physical_plan::operator_statistics::DefaultStatisticsProvider` is
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org