Move newlines_in_values from FileScanConfig to CsvSource

This PR moves the CSV-specific `newlines_in_values` configuration
option from `FileScanConfig` (a shared format-agnostic configuration)
to `CsvSource` where it belongs.

Changes:
- Add `newlines_in_values` field and methods to `CsvSource`
- Add `has_newlines_in_values()` method to `FileSource` trait
- Update `FileSource::repartitioned()` to use the trait method
- Remove `new_lines_in_values` from `FileScanConfig` and builder
- Update proto serialization to use `CsvSource`
- Update tests and documentation

Closes apache#18453

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

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

Author: adriangb

datafusion/datasource-arrow/src/source.rs

@@ -460,9 +460,7 @@ impl FileSource for ArrowSource {
                 // Use the default trait implementation logic for file format
                 use datafusion_datasource::file_groups::FileGroupPartitioner;
 
-                if config.file_compression_type.is_compressed()
-                    || config.new_lines_in_values
-                {
+                if config.file_compression_type.is_compressed() {
                     return Ok(None);
                 }
 

datafusion/datasource-csv/src/file_format.rs

@@ -445,11 +445,15 @@ 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));
+        let source = Arc::new(
+            csv_source
+                .clone()
+                .with_csv_options(csv_options)
+                .with_newlines_in_values(newlines_in_values),
+        );
 
         let config = FileScanConfigBuilder::from(conf)
             .with_file_compression_type(self.options.compression.into())
-            .with_newlines_in_values(newlines_in_values)
             .with_source(source)
             .build();
 

datafusion/datasource-csv/src/source.rs

@@ -77,11 +77,11 @@ use tokio::io::AsyncWriteExt;
 /// 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)
 ///     .with_file(PartitionedFile::new("file1.csv", 100*1024*1024))
-///     .with_newlines_in_values(true) // The file contains newlines in values;
 ///     .build();
 /// let exec = (DataSourceExec::from_data_source(config));
 /// ```
@@ -93,6 +93,8 @@ 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 {
@@ -106,6 +108,7 @@ impl CsvSource {
             batch_size: None,
             metrics: ExecutionPlanMetricsSet::new(),
             schema_adapter_factory: None,
+            newlines_in_values: false,
         }
     }
 
@@ -176,6 +179,21 @@ impl CsvSource {
         conf.options.truncated_rows = Some(truncate_rows);
         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.
+    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
+    }
 }
 
 impl CsvSource {
@@ -297,6 +315,11 @@ impl FileSource for CsvSource {
     fn file_type(&self) -> &str {
         "csv"
     }
+
+    fn has_newlines_in_values(&self) -> bool {
+        self.newlines_in_values
+    }
+
     fn fmt_extra(&self, t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {
         match t {
             DisplayFormatType::Default | DisplayFormatType::Verbose => {

datafusion/datasource/src/file.rs

@@ -84,6 +84,19 @@ pub trait FileSource: Send + Sync {
         Ok(())
     }
 
+    /// Returns whether this file source has values that may contain newline characters.
+    ///
+    /// 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.
+    ///
+    /// 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
+    }
+
     /// If supported by the [`FileSource`], redistribute files across partitions
     /// according to their size. Allows custom file formats to implement their
     /// own repartitioning logic.
@@ -97,7 +110,7 @@ pub trait FileSource: Send + Sync {
         output_ordering: Option<LexOrdering>,
         config: &FileScanConfig,
     ) -> Result<Option<FileScanConfig>> {
-        if config.file_compression_type.is_compressed() || config.new_lines_in_values {
+        if config.file_compression_type.is_compressed() || self.has_newlines_in_values() {
             return Ok(None);
         }
 

datafusion/datasource/src/file_scan_config.rs

@@ -159,8 +159,6 @@ pub struct FileScanConfig {
     pub output_ordering: Vec<LexOrdering>,
     /// File compression type
     pub file_compression_type: FileCompressionType,
-    /// Are new lines in values supported for CSVOptions
-    pub new_lines_in_values: bool,
     /// File source such as `ParquetSource`, `CsvSource`, `JsonSource`, etc.
     pub file_source: Arc<dyn FileSource>,
     /// Batch size while creating new batches
@@ -250,7 +248,6 @@ pub struct FileScanConfigBuilder {
     statistics: Option<Statistics>,
     output_ordering: Vec<LexOrdering>,
     file_compression_type: Option<FileCompressionType>,
-    new_lines_in_values: Option<bool>,
     batch_size: Option<usize>,
     expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
     partitioned_by_file_group: bool,
@@ -274,7 +271,6 @@ impl FileScanConfigBuilder {
             statistics: None,
             output_ordering: vec![],
             file_compression_type: None,
-            new_lines_in_values: None,
             limit: None,
             constraints: None,
             batch_size: None,
@@ -413,16 +409,6 @@ impl FileScanConfigBuilder {
         self
     }
 
-    /// Set whether new lines in values are supported for CSVOptions
-    ///
-    /// Parsing newlines in quoted values may be affected by execution behaviour such as
-    /// parallel file scanning. Setting this to `true` ensures that newlines in values are
-    /// parsed successfully, which may reduce performance.
-    pub fn with_newlines_in_values(mut self, new_lines_in_values: bool) -> Self {
-        self.new_lines_in_values = Some(new_lines_in_values);
-        self
-    }
-
     /// Set the batch_size property
     pub fn with_batch_size(mut self, batch_size: Option<usize>) -> Self {
         self.batch_size = batch_size;
@@ -472,7 +458,6 @@ impl FileScanConfigBuilder {
             statistics,
             output_ordering,
             file_compression_type,
-            new_lines_in_values,
             batch_size,
             expr_adapter_factory: expr_adapter,
             partitioned_by_file_group,
@@ -484,7 +469,6 @@ impl FileScanConfigBuilder {
         });
         let file_compression_type =
             file_compression_type.unwrap_or(FileCompressionType::UNCOMPRESSED);
-        let new_lines_in_values = new_lines_in_values.unwrap_or(false);
 
         FileScanConfig {
             object_store_url,
@@ -494,7 +478,6 @@ impl FileScanConfigBuilder {
             file_groups,
             output_ordering,
             file_compression_type,
-            new_lines_in_values,
             batch_size,
             expr_adapter_factory: expr_adapter,
             statistics,
@@ -512,7 +495,6 @@ impl From<FileScanConfig> for FileScanConfigBuilder {
             statistics: Some(config.statistics),
             output_ordering: config.output_ordering,
             file_compression_type: Some(config.file_compression_type),
-            new_lines_in_values: Some(config.new_lines_in_values),
             limit: config.limit,
             constraints: Some(config.constraints),
             batch_size: config.batch_size,
@@ -913,17 +895,6 @@ impl FileScanConfig {
         props.constraints().clone()
     }
 
-    /// Specifies whether newlines in (quoted) values are supported.
-    ///
-    /// Parsing newlines in quoted values may be affected by execution behaviour such as
-    /// parallel file scanning. Setting this to `true` ensures that newlines in values are
-    /// parsed successfully, which may reduce performance.
-    ///
-    /// The default behaviour depends on the `datafusion.catalog.newlines_in_values` setting.
-    pub fn newlines_in_values(&self) -> bool {
-        self.new_lines_in_values
-    }
-
     #[deprecated(
         since = "52.0.0",
         note = "This method is no longer used, use eq_properties instead. It will be removed in 58.0.0 or 6 months after 52.0.0 is released, whichever comes first."
@@ -1722,7 +1693,6 @@ mod tests {
                 .into(),
             ])
             .with_file_compression_type(FileCompressionType::UNCOMPRESSED)
-            .with_newlines_in_values(true)
             .build();
 
         // Verify the built config has all the expected values
@@ -1749,7 +1719,6 @@ mod tests {
             config.file_compression_type,
             FileCompressionType::UNCOMPRESSED
         );
-        assert!(config.new_lines_in_values);
         assert_eq!(config.output_ordering.len(), 1);
     }
 
@@ -1844,7 +1813,6 @@ mod tests {
             config.file_compression_type,
             FileCompressionType::UNCOMPRESSED
         );
-        assert!(!config.new_lines_in_values);
         assert!(config.output_ordering.is_empty());
         assert!(config.constraints.is_empty());
 
@@ -1892,7 +1860,6 @@ mod tests {
         .with_limit(Some(10))
         .with_file(file.clone())
         .with_constraints(Constraints::default())
-        .with_newlines_in_values(true)
         .build();
 
         // Create a new builder from the config
@@ -1922,7 +1889,6 @@ mod tests {
             "test_file.parquet"
         );
         assert_eq!(new_config.constraints, Constraints::default());
-        assert!(new_config.new_lines_in_values);
     }
 
     #[test]

datafusion/proto/src/physical_plan/mod.rs

@@ -643,13 +643,21 @@ 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,
             extension_codec,
             source,
         )?)
-        .with_newlines_in_values(scan.newlines_in_values)
         .with_file_compression_type(FileCompressionType::UNCOMPRESSED)
         .build();
         Ok(DataSourceExec::from_data_source(conf))
@@ -2628,7 +2636,7 @@ impl protobuf::PhysicalPlanNode {
                             } else {
                                 None
                             },
-                            newlines_in_values: maybe_csv.newlines_in_values(),
+                            newlines_in_values: csv_config.newlines_in_values(),
                             truncate_rows: csv_config.truncate_rows(),
                         },
                     )),

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -933,7 +933,6 @@ async fn roundtrip_parquet_exec_with_table_partition_cols() -> Result<()> {
         FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source)
             .with_projection_indices(Some(vec![0, 1]))?
             .with_file_group(FileGroup::new(vec![file_group]))
-            .with_newlines_in_values(false)
             .build();
 
     roundtrip_test(DataSourceExec::from_data_source(scan_config))

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

@@ -57,6 +57,36 @@ 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`
+
+The CSV-specific `newlines_in_values` configuration option has been moved from `FileScanConfig` to `CsvSource`, 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`:
+
+**Before:**
+
+```rust,ignore
+let source = Arc::new(CsvSource::new(file_schema.clone()));
+let config = FileScanConfigBuilder::new(object_store_url, source)
+    .with_newlines_in_values(true)
+    .build();
+```
+
+**After:**
+
+```rust,ignore
+let source = Arc::new(CsvSource::new(file_schema.clone())
+    .with_newlines_in_values(true));
+let config = FileScanConfigBuilder::new(object_store_url, source)
+    .build();
+```
+
 ### Removal of `pyarrow` feature
 
 The `pyarrow` feature flag has been removed. This feature has been migrated to