Add quote style and trimming to csv writier (#20813)

## Which issue does this PR close?

- Closes #10669

Related arrow-rs PRs apache/arrow-rs#8960 and
apache/arrow-rs#9004

## Rationale for this change

The CSV writer was missing support for `quote_style`,
`ignore_leading_whitespace`, and `ignore_trailing_whitespace` options
that are available on the underlying arrow `WriterBuilder`. This meant
users couldn't control quoting behaviour or whitespace trimming when
writing CSV files.

## What changes are included in this PR?

Adds three new CSV writer options wired through the full stack:

- `quote_style` — controls when fields are quoted (`Always`,
`Necessary`, `NonNumeric`, `Never`). Modelled as a protobuf enum
(`CsvQuoteStyle`).
- `ignore_leading_whitespace` — trims leading whitespace from string
values on write.
- `ignore_trailing_whitespace` — trims trailing whitespace from string
values on write.

## Are these changes tested?

Yes — sqllogictest coverage added in `csv_files.slt`

## Are there any user-facing changes?

Three new `format.*` options available in COPY TO and CREATE EXTERNAL
TABLE for CSV:
- `format.quote_style` (string: `Always`, `Necessary`, `NonNumeric`,
`Never`)
- `format.ignore_leading_whitespace` (boolean)
- `format.ignore_trailing_whitespace` (boolean)

Author: xanderbailey

datafusion/common/src/config.rs

@@ -24,7 +24,7 @@ use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
 use crate::error::_config_err;
 use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
 use crate::parquet_config::DFParquetWriterVersion;
-use crate::parsers::CompressionTypeVariant;
+use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
 use crate::utils::get_available_parallelism;
 use crate::{DataFusionError, Result};
 #[cfg(feature = "parquet_encryption")]
@@ -2042,6 +2042,17 @@ impl ConfigField for CompressionTypeVariant {
     }
 }
 
+impl ConfigField for CsvQuoteStyle {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, _: &str, value: &str) -> Result<()> {
+        *self = CsvQuoteStyle::from_str(value)?;
+        Ok(())
+    }
+}
+
 /// An implementation trait used to recursively walk configuration
 pub trait Visit {
     fn some<V: Display>(&mut self, key: &str, value: V, description: &'static str);
@@ -3114,6 +3125,15 @@ config_namespace! {
         pub terminator: Option<u8>, default = None
         pub escape: Option<u8>, default = None
         pub double_quote: Option<bool>, default = None
+        /// Quote style for CSV writing.
+        /// One of: "Always", "Necessary", "NonNumeric", "Never"
+        pub quote_style: CsvQuoteStyle, default = CsvQuoteStyle::Necessary
+        /// Whether to ignore leading whitespace in string values when writing CSV.
+        /// Defaults to `false` when `None`.
+        pub ignore_leading_whitespace: Option<bool>, default = None
+        /// Whether to ignore trailing whitespace in string values when writing CSV.
+        /// Defaults to `false` when `None`.
+        pub ignore_trailing_whitespace: Option<bool>, default = None
         /// Specifies whether newlines in (quoted) values are supported.
         ///
         /// Parsing newlines in quoted values may be affected by execution behaviour such as
@@ -3222,6 +3242,30 @@ impl CsvOptions {
         self
     }
 
+    /// Set the quote style for CSV writing.
+    pub fn with_quote_style(mut self, quote_style: CsvQuoteStyle) -> Self {
+        self.quote_style = quote_style;
+        self
+    }
+
+    /// Set whether to ignore leading whitespace in string values when writing CSV.
+    pub fn with_ignore_leading_whitespace(
+        mut self,
+        ignore_leading_whitespace: bool,
+    ) -> Self {
+        self.ignore_leading_whitespace = Some(ignore_leading_whitespace);
+        self
+    }
+
+    /// Set whether to ignore trailing whitespace in string values when writing CSV.
+    pub fn with_ignore_trailing_whitespace(
+        mut self,
+        ignore_trailing_whitespace: bool,
+    ) -> Self {
+        self.ignore_trailing_whitespace = Some(ignore_trailing_whitespace);
+        self
+    }
+
     /// Specifies whether newlines in (quoted) values are supported.
     ///
     /// Parsing newlines in quoted values may be affected by execution behaviour such as

datafusion/common/src/file_options/csv_writer.rs

@@ -94,6 +94,13 @@ impl TryFrom<&CsvOptions> for CsvWriterOptions {
         if let Some(v) = &value.double_quote {
             builder = builder.with_double_quote(*v)
         }
+        builder = builder.with_quote_style(value.quote_style.into());
+        if let Some(v) = &value.ignore_leading_whitespace {
+            builder = builder.with_ignore_leading_whitespace(*v)
+        }
+        if let Some(v) = &value.ignore_trailing_whitespace {
+            builder = builder.with_ignore_trailing_whitespace(*v)
+        }
         Ok(CsvWriterOptions {
             writer_options: builder,
             compression: value.compression,

datafusion/common/src/parsers.rs

@@ -73,3 +73,59 @@ impl CompressionTypeVariant {
         !matches!(self, &Self::UNCOMPRESSED)
     }
 }
+
+/// CSV quote style
+///
+/// Controls when fields are quoted when writing CSV files.
+/// Corresponds to [`arrow::csv::QuoteStyle`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
+pub enum CsvQuoteStyle {
+    /// Quote all fields
+    Always,
+    /// Only quote fields when necessary (default)
+    #[default]
+    Necessary,
+    /// Quote all non-numeric fields
+    NonNumeric,
+    /// Never quote fields
+    Never,
+}
+
+impl FromStr for CsvQuoteStyle {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "always" => Ok(Self::Always),
+            "necessary" => Ok(Self::Necessary),
+            "non_numeric" | "nonnumeric" => Ok(Self::NonNumeric),
+            "never" => Ok(Self::Never),
+            _ => Err(DataFusionError::NotImplemented(format!(
+                "Unsupported CSV quote style {s}"
+            ))),
+        }
+    }
+}
+
+impl From<CsvQuoteStyle> for arrow::csv::QuoteStyle {
+    fn from(style: CsvQuoteStyle) -> Self {
+        match style {
+            CsvQuoteStyle::Always => Self::Always,
+            CsvQuoteStyle::NonNumeric => Self::NonNumeric,
+            CsvQuoteStyle::Never => Self::Never,
+            CsvQuoteStyle::Necessary => Self::Necessary,
+        }
+    }
+}
+
+impl Display for CsvQuoteStyle {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let str = match self {
+            Self::Always => "Always",
+            Self::Necessary => "Necessary",
+            Self::NonNumeric => "NonNumeric",
+            Self::Never => "Never",
+        };
+        write!(f, "{str}")
+    }
+}

datafusion/proto-common/proto/datafusion_common.proto

@@ -434,6 +434,13 @@ message JsonWriterOptions {
 }
 
 
+enum CsvQuoteStyle {
+  NECESSARY = 0;
+  ALWAYS = 1;
+  NON_NUMERIC = 2;
+  NEVER = 3;
+}
+
 message CsvWriterOptions {
   // Compression type
   CompressionTypeVariant compression = 1;
@@ -457,6 +464,12 @@ message CsvWriterOptions {
   string escape = 10;
   // Optional flag whether to double quotes, instead of escaping. Defaults to `true`
   bool double_quote = 11;
+  // Quote style for CSV writing
+  CsvQuoteStyle quote_style = 12;
+  // Whether to ignore leading whitespace in string values
+  bool ignore_leading_whitespace = 13;
+  // Whether to ignore trailing whitespace in string values
+  bool ignore_trailing_whitespace = 14;
 }
 
 // Options controlling CSV format
@@ -480,6 +493,12 @@ message CsvOptions {
   bytes terminator = 17; // Optional terminator character as a byte
   bytes truncated_rows = 18; // Indicates if truncated rows are allowed
   optional uint32 compression_level = 19; // Optional compression level
+  // Quote style for CSV writing
+  CsvQuoteStyle quote_style = 20;
+  // Whether to ignore leading whitespace in string values
+  bytes ignore_leading_whitespace = 21;
+  // Whether to ignore trailing whitespace in string values
+  bytes ignore_trailing_whitespace = 22;
 }
 
 // Options controlling CSV format

datafusion/proto-common/src/from_proto/mod.rs

@@ -22,7 +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::WriterBuilder;
+use arrow::csv::{QuoteStyle, WriterBuilder};
 use arrow::datatypes::{
     DataType, Field, IntervalDayTimeType, IntervalMonthDayNanoType, IntervalUnit, Schema,
     TimeUnit, UnionFields, UnionMode, i256,
@@ -964,6 +964,17 @@ impl From<CompressionTypeVariant> for protobuf::CompressionTypeVariant {
     }
 }
 
+impl From<protobuf::CsvQuoteStyle> for datafusion_common::parsers::CsvQuoteStyle {
+    fn from(value: protobuf::CsvQuoteStyle) -> Self {
+        match value {
+            protobuf::CsvQuoteStyle::Necessary => Self::Necessary,
+            protobuf::CsvQuoteStyle::Always => Self::Always,
+            protobuf::CsvQuoteStyle::NonNumeric => Self::NonNumeric,
+            protobuf::CsvQuoteStyle::Never => Self::Never,
+        }
+    }
+}
+
 impl TryFrom<&protobuf::CsvWriterOptions> for CsvWriterOptions {
     type Error = DataFusionError;
 
@@ -1020,6 +1031,15 @@ impl TryFrom<&protobuf::CsvOptions> for CsvOptions {
                 .then(|| proto_opts.null_regex.clone()),
             comment: proto_opts.comment.first().copied(),
             truncated_rows: proto_opts.truncated_rows.first().map(|h| *h != 0),
+            quote_style: proto_opts.quote_style().into(),
+            ignore_leading_whitespace: proto_opts
+                .ignore_leading_whitespace
+                .first()
+                .map(|h| *h != 0),
+            ignore_trailing_whitespace: proto_opts
+                .ignore_trailing_whitespace
+                .first()
+                .map(|h| *h != 0),
         })
     }
 }
@@ -1281,14 +1301,27 @@ pub(crate) fn csv_writer_options_from_proto(
             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_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)]