Deprecate CsvSource::newlines_in_values and preserve proto coverage (#26179)

## Which issue does this PR close?

- Part of #24612

## Rationale for this change

Deprecate `CsvSource::newlines_in_values()` as part of the CSV API
cleanup while preserving compatibility and existing CSV serialization
behavior.

The getter simply returns
`CsvOptions.newlines_in_values.unwrap_or(false)`. There is no
replacement getter, so callers should retain their original `CsvOptions`
when they need to inspect this setting.

The public getter remains callable during the deprecation window, and
CSV reading and repartitioning behavior are unchanged.

## What changes are included in this PR?

- Mark `CsvSource::newlines_in_values()` as deprecated starting in
56.0.0, with an accurate migration note.
- Replace the internal encoder's getter call with
`self.options.newlines_in_values.unwrap_or(false)`, preserving the
existing boolean mapping.
- Extend `roundtrip_csv_scan_preserves_format_options` to cover
`Some(true)`, `Some(false)`, and `None`.
- Assert the protobuf `CsvScan.newlines_in_values` field after
round-tripping the physical plan, while retaining all existing CSV
format-option assertions.
- Preserve malformed terminator validation for empty and two-byte
terminators.
- Document the deprecation and migration guidance in the 56.0.0 upgrade
guide.

No protobuf wire-format changes or new public APIs are introduced.

## Are these changes tested?

Extended `roundtrip_csv_scan_preserves_format_options` to verify:

- `Some(true)` serializes as `true`.
- `Some(false)` serializes as `false`.
- `None` serializes as `false`.
- Existing CSV format options, including delimiter, quote, escape,
comment, terminator, truncated rows, header, and GZIP compression,
survive the round-trip.
- Invalid zero-byte and two-byte terminators continue to produce
decoding errors.

The patch includes these regression tests, but does not establish
whether they have been executed successfully.

Recommended verification commands:

```bash
cargo test -p datafusion-proto --test proto_integration roundtrip_csv_scan_preserves_format_options
cargo test -p datafusion-datasource-csv
cargo fmt --all
cargo clippy --all-targets --all-features -- -D warnings
./ci/scripts/doc_prettier_check.sh --write --allow-dirty
```

## Are there any user-facing changes?

`CsvSource::newlines_in_values()` is deprecated starting in 56.0.0 but
remains callable with unchanged behavior.

Existing callers may receive deprecation warnings, potentially affecting
downstream builds that treat warnings as errors.

There is no replacement getter. Callers should retain the `CsvOptions`
used to configure the source and read
`options.newlines_in_values.unwrap_or(false)`.

The 56.0.0 upgrade guide documents this migration.

CSV parsing, repartitioning, and protobuf serialization behavior are
unchanged.

## LLM-generated code disclosure

This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed.

Author: kosiew

datafusion/datasource-csv/src/source.rs

@@ -174,6 +174,10 @@ impl CsvSource {
     }
 
     /// Whether values may contain newline characters
+    #[deprecated(
+        since = "56.0.0",
+        note = "No replacement getter exists. Retain CsvOptions and read its newlines_in_values field with unwrap_or(false)."
+    )]
     pub fn newlines_in_values(&self) -> bool {
         self.options.newlines_in_values.unwrap_or(false)
     }
@@ -354,7 +358,7 @@ impl FileSource for CsvSource {
                     )
                 })
                 .transpose()?,
-            newlines_in_values: self.newlines_in_values(),
+            newlines_in_values: self.options.newlines_in_values.unwrap_or(false),
             truncate_rows: self.truncate_rows(),
             terminator: self.terminator().map(|terminator| vec![terminator]),
         };

datafusion/proto/tests/cases/plans/sources.rs

@@ -535,76 +535,89 @@ fn roundtrip_csv_scan_preserves_format_options() -> Result<()> {
 
     let file_schema =
         Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)]));
-    let table_schema = TableSchema::from(&file_schema);
-    let file_source =
-        Arc::new(CsvSource::new(table_schema).with_csv_options(CsvOptions {
-            has_header: Some(false),
-            delimiter: b'|',
-            quote: b'\'',
-            escape: Some(b'\\'),
-            comment: Some(b'#'),
-            terminator: Some(0xff),
-            newlines_in_values: Some(true),
-            truncated_rows: Some(true),
-            ..Default::default()
-        }));
-
-    let scan_config =
-        FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source)
-            .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::new(
-                "/path/to/file.csv.gz".to_string(),
-                1024,
-            )])])
-            .with_file_compression_type(FileCompressionType::GZIP)
-            .build();
-
-    let ctx = SessionContext::new();
-    let codec = DefaultPhysicalExtensionCodec {};
-    let plan: Arc<dyn ExecutionPlan> = DataSourceExec::from_data_source(scan_config);
-    let roundtripped = roundtrip_test_and_return(
-        Arc::clone(&plan),
-        &ctx,
-        &codec,
-        &DefaultPhysicalProtoConverter {},
-    )?;
-    let data_source = roundtripped
-        .downcast_ref::<DataSourceExec>()
-        .ok_or_else(|| internal_datafusion_err!("Expected DataSourceExec"))?;
-    let file_scan = data_source
-        .data_source()
-        .downcast_ref::<FileScanConfig>()
-        .ok_or_else(|| internal_datafusion_err!("Expected FileScanConfig"))?;
-    let csv_source = file_scan
-        .file_source()
-        .downcast_ref::<CsvSource>()
-        .ok_or_else(|| internal_datafusion_err!("Expected CsvSource"))?;
-
-    assert!(!csv_source.has_header());
-    assert_eq!(csv_source.delimiter(), b'|');
-    assert_eq!(csv_source.quote(), b'\'');
-    assert_eq!(csv_source.escape(), Some(b'\\'));
-    assert_eq!(csv_source.comment(), Some(b'#'));
-    assert_eq!(csv_source.terminator(), Some(0xff));
-    assert!(csv_source.newlines_in_values());
-    assert!(csv_source.truncate_rows());
-    assert_eq!(file_scan.file_compression_type, FileCompressionType::GZIP);
+    for newlines_in_values in [Some(true), Some(false), None] {
+        let table_schema = TableSchema::from(&file_schema);
+        let file_source =
+            Arc::new(CsvSource::new(table_schema).with_csv_options(CsvOptions {
+                has_header: Some(false),
+                delimiter: b'|',
+                quote: b'\'',
+                escape: Some(b'\\'),
+                comment: Some(b'#'),
+                terminator: Some(0xff),
+                newlines_in_values,
+                truncated_rows: Some(true),
+                ..Default::default()
+            }));
+
+        let scan_config =
+            FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source)
+                .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::new(
+                    "/path/to/file.csv.gz".to_string(),
+                    1024,
+                )])])
+                .with_file_compression_type(FileCompressionType::GZIP)
+                .build();
 
-    for invalid_terminator in [vec![], vec![b'\r', b'\n']] {
-        let mut node =
-            PhysicalPlanNode::try_from_physical_plan(Arc::clone(&plan), &codec)?;
-        match node.physical_plan_type.as_mut() {
+        let ctx = SessionContext::new();
+        let codec = DefaultPhysicalExtensionCodec {};
+        let plan: Arc<dyn ExecutionPlan> = DataSourceExec::from_data_source(scan_config);
+        let roundtripped = roundtrip_test_and_return(
+            Arc::clone(&plan),
+            &ctx,
+            &codec,
+            &DefaultPhysicalProtoConverter {},
+        )?;
+        let data_source = roundtripped
+            .downcast_ref::<DataSourceExec>()
+            .ok_or_else(|| internal_datafusion_err!("Expected DataSourceExec"))?;
+        let file_scan = data_source
+            .data_source()
+            .downcast_ref::<FileScanConfig>()
+            .ok_or_else(|| internal_datafusion_err!("Expected FileScanConfig"))?;
+        let csv_source = file_scan
+            .file_source()
+            .downcast_ref::<CsvSource>()
+            .ok_or_else(|| internal_datafusion_err!("Expected CsvSource"))?;
+
+        assert!(!csv_source.has_header());
+        assert_eq!(csv_source.delimiter(), b'|');
+        assert_eq!(csv_source.quote(), b'\'');
+        assert_eq!(csv_source.escape(), Some(b'\\'));
+        assert_eq!(csv_source.comment(), Some(b'#'));
+        assert_eq!(csv_source.terminator(), Some(0xff));
+        let node =
+            PhysicalPlanNode::try_from_physical_plan(Arc::clone(&roundtripped), &codec)?;
+        match node.physical_plan_type {
             Some(protobuf::physical_plan_node::PhysicalPlanType::CsvScan(scan)) => {
-                scan.terminator = Some(invalid_terminator);
+                assert_eq!(scan.newlines_in_values, newlines_in_values.unwrap_or(false));
             }
             other => return internal_err!("Expected CsvScan node, got {other:?}"),
         }
-        let err = node
-            .try_into_physical_plan(ctx.task_ctx().as_ref(), &codec)
-            .expect_err("invalid terminator length must fail");
-        assert!(
-            err.to_string().contains("expected exactly one byte"),
-            "unexpected error: {err}"
-        );
+        assert!(csv_source.truncate_rows());
+        assert_eq!(file_scan.file_compression_type, FileCompressionType::GZIP);
+
+        // Terminator validation is independent of the newline setting.
+        if newlines_in_values != Some(true) {
+            continue;
+        }
+        for invalid_terminator in [vec![], vec![b'\r', b'\n']] {
+            let mut node =
+                PhysicalPlanNode::try_from_physical_plan(Arc::clone(&plan), &codec)?;
+            match node.physical_plan_type.as_mut() {
+                Some(protobuf::physical_plan_node::PhysicalPlanType::CsvScan(scan)) => {
+                    scan.terminator = Some(invalid_terminator);
+                }
+                other => return internal_err!("Expected CsvScan node, got {other:?}"),
+            }
+            let err = node
+                .try_into_physical_plan(ctx.task_ctx().as_ref(), &codec)
+                .expect_err("invalid terminator length must fail");
+            assert!(
+                err.to_string().contains("expected exactly one byte"),
+                "unexpected error: {err}"
+            );
+        }
     }
     Ok(())
 }

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

@@ -115,6 +115,20 @@ change. Use `FieldMetadata::to_hashmap` where a `HashMap` is still required.
 
 [`metadata`]: https://docs.rs/arrow/60.0.0/arrow/datatypes/struct.Metadata.html
 
+### `CsvSource::newlines_in_values` is deprecated
+
+`CsvSource::newlines_in_values()` is deprecated starting in 56.0.0 but remains
+callable with unchanged behavior. There is no replacement getter, and
+`CsvSource` does not expose a `csv_options()` method.
+
+**Migration guide:**
+
+Retain the `CsvOptions` used to configure the source before passing them to
+`CsvSource::with_csv_options`. Read `options.newlines_in_values.unwrap_or(false)`
+to obtain the same boolean value: `None` and `Some(false)` mean false, while
+`Some(true)` means true. CSV reading, repartitioning, and serialization behavior
+are unchanged.
+
 ### `ParquetAccessPlan::into_overall_row_selection` is deprecated
 
 `ParquetAccessPlan::into_overall_row_selection` is deprecated starting in 56.0.0.