Address PR feedback: use CsvOptions and supports_repartitioning()

- Remove separate `newlines_in_values` field from `CsvSource`
- Use `CsvOptions.newlines_in_values` instead (already exists)
- Rename `has_newlines_in_values()` to `supports_repartitioning()` in
  `FileSource` trait - describes capability, not CSV-specific reason
- `CsvSource` overrides `supports_repartitioning()` to check options
- Update proto serialization to set value in `CsvOptions`
- Update documentation and migration guide

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: adriangb

datafusion/datasource-csv/src/file_format.rs

@@ -437,6 +437,7 @@ impl FileFormat for CsvFormat {
 
         let mut csv_options = self.options.clone();
         csv_options.has_header = Some(has_header);
+        csv_options.newlines_in_values = Some(newlines_in_values);
 
         // Get the existing CsvSource and update its options
         // We need to preserve the table_schema from the original source (which includes partition columns)
@@ -445,12 +446,7 @@ impl FileFormat for CsvFormat {
             .as_any()
             .downcast_ref::<CsvSource>()
             .expect("file_source should be a CsvSource");
-        let source = Arc::new(
-            csv_source
-                .clone()
-                .with_csv_options(csv_options)
-                .with_newlines_in_values(newlines_in_values),
-        );
+        let source = Arc::new(csv_source.clone().with_csv_options(csv_options));
 
         let config = FileScanConfigBuilder::from(conf)
             .with_file_compression_type(self.options.compression.into())

datafusion/datasource-csv/src/source.rs

@@ -72,12 +72,12 @@ use tokio::io::AsyncWriteExt;
 ///     has_header: Some(true),
 ///     delimiter: b',',
 ///     quote: b'"',
+///     newlines_in_values: Some(true), // The file contains newlines in values
 ///     ..Default::default()
 /// };
 /// let source = Arc::new(CsvSource::new(file_schema.clone())
 ///     .with_csv_options(options)
 ///     .with_terminator(Some(b'#'))
-///     .with_newlines_in_values(true) // The file contains newlines in values
 /// );
 /// // Create a DataSourceExec for reading the first 100MB of `file1.csv`
 /// let config = FileScanConfigBuilder::new(object_store_url, source)
@@ -93,8 +93,6 @@ pub struct CsvSource {
     projection: SplitProjection,
     metrics: ExecutionPlanMetricsSet,
     schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
-    /// Whether values may contain newline characters
-    newlines_in_values: bool,
 }
 
 impl CsvSource {
@@ -108,7 +106,6 @@ impl CsvSource {
             batch_size: None,
             metrics: ExecutionPlanMetricsSet::new(),
             schema_adapter_factory: None,
-            newlines_in_values: false,
         }
     }
 
@@ -180,19 +177,9 @@ impl CsvSource {
         conf
     }
 
-    /// Whether values may contain newline characters.
-    ///
-    /// When enabled, scanning cannot be parallelized across a single file
-    /// because newlines in values prevent determining record boundaries
-    /// by byte offset alone.
+    /// Whether values may contain newline characters
     pub fn newlines_in_values(&self) -> bool {
-        self.newlines_in_values
-    }
-
-    /// Set whether values may contain newline characters
-    pub fn with_newlines_in_values(mut self, newlines_in_values: bool) -> Self {
-        self.newlines_in_values = newlines_in_values;
-        self
+        self.options.newlines_in_values.unwrap_or(false)
     }
 }
 
@@ -316,8 +303,10 @@ impl FileSource for CsvSource {
         "csv"
     }
 
-    fn has_newlines_in_values(&self) -> bool {
-        self.newlines_in_values
+    fn supports_repartitioning(&self) -> bool {
+        // Cannot repartition if values may contain newlines, as record
+        // boundaries cannot be determined by byte offset alone
+        !self.options.newlines_in_values.unwrap_or(false)
     }
 
     fn fmt_extra(&self, t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {

datafusion/datasource/src/file.rs

@@ -84,17 +84,19 @@ pub trait FileSource: Send + Sync {
         Ok(())
     }
 
-    /// Returns whether this file source has values that may contain newline characters.
+    /// Returns whether this file source supports repartitioning files by byte ranges.
     ///
-    /// This is primarily relevant for CSV files where quoted values can contain
-    /// embedded newlines. When this returns `true`, files cannot be repartitioned
-    /// by byte ranges because record boundaries cannot be determined by simple
-    /// newline scanning.
+    /// When this returns `true`, files can be split into multiple partitions
+    /// based on byte offsets for parallel reading.
     ///
-    /// The default implementation returns `false`. CSV sources should override
-    /// this method to return the appropriate value based on their configuration.
-    fn has_newlines_in_values(&self) -> bool {
-        false
+    /// When this returns `false`, files cannot be repartitioned (e.g., CSV files
+    /// with `newlines_in_values` enabled cannot be split because record boundaries
+    /// cannot be determined by byte offset alone).
+    ///
+    /// The default implementation returns `true`. File sources that cannot support
+    /// repartitioning should override this method.
+    fn supports_repartitioning(&self) -> bool {
+        true
     }
 
     /// If supported by the [`FileSource`], redistribute files across partitions
@@ -110,7 +112,7 @@ pub trait FileSource: Send + Sync {
         output_ordering: Option<LexOrdering>,
         config: &FileScanConfig,
     ) -> Result<Option<FileScanConfig>> {
-        if config.file_compression_type.is_compressed() || self.has_newlines_in_values() {
+        if config.file_compression_type.is_compressed() || !self.supports_repartitioning() {
             return Ok(None);
         }
 

datafusion/proto/src/physical_plan/mod.rs

@@ -634,6 +634,7 @@ impl protobuf::PhysicalPlanNode {
             has_header: Some(scan.has_header),
             delimiter: str_to_byte(&scan.delimiter, "delimiter")?,
             quote: str_to_byte(&scan.quote, "quote")?,
+            newlines_in_values: Some(scan.newlines_in_values),
             ..Default::default()
         };
         let source = Arc::new(
@@ -643,15 +644,6 @@ impl protobuf::PhysicalPlanNode {
                 .with_comment(comment),
         );
 
-        let source = Arc::new(
-            source
-                .as_any()
-                .downcast_ref::<CsvSource>()
-                .expect("source must be CsvSource")
-                .clone()
-                .with_newlines_in_values(scan.newlines_in_values),
-        );
-
         let conf = FileScanConfigBuilder::from(parse_protobuf_file_scan_config(
             scan.base_conf.as_ref().unwrap(),
             ctx,

docs/source/library-user-guide/upgrading.md

@@ -57,17 +57,17 @@ See <https://github.com/apache/datafusion/issues/19056> for more details.
 
 Note that the internal API has changed to use a trait `ListFilesCache` instead of a type alias.
 
-### `newlines_in_values` moved from `FileScanConfig` to `CsvSource`
+### `newlines_in_values` moved from `FileScanConfig` to `CsvOptions`
 
-The CSV-specific `newlines_in_values` configuration option has been moved from `FileScanConfig` to `CsvSource`, as it only applies to CSV file parsing.
+The CSV-specific `newlines_in_values` configuration option has been moved from `FileScanConfig` to `CsvOptions`, as it only applies to CSV file parsing.
 
 **Who is affected:**
 
 - Users who set `newlines_in_values` via `FileScanConfigBuilder::with_newlines_in_values()`
 
 **Migration guide:**
 
-Set `newlines_in_values` on `CsvSource` instead of `FileScanConfigBuilder`:
+Set `newlines_in_values` in `CsvOptions` instead of on `FileScanConfigBuilder`:
 
 **Before:**
 
@@ -81,8 +81,12 @@ let config = FileScanConfigBuilder::new(object_store_url, source)
 **After:**
 
 ```rust,ignore
+let options = CsvOptions {
+    newlines_in_values: Some(true),
+    ..Default::default()
+};
 let source = Arc::new(CsvSource::new(file_schema.clone())
-    .with_newlines_in_values(true));
+    .with_csv_options(options));
 let config = FileScanConfigBuilder::new(object_store_url, source)
     .build();
 ```