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/10/11 09:12:33 UTC
(datafusion) 01/01: Deprecate CsvSource::newlines_in_values and preserve proto coverage (#26179)
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-26179-a4b8198cfa2419394df966f815d284e745f06653
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 75ddead2b765ff6c574c4cc989538fabcf250c9f
Author: kosiew <ko...@gmail.com>
AuthorDate: Sun Oct 11 09:12:07 2026 +0000
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.
---
datafusion/datasource-csv/src/source.rs | 6 +-
datafusion/proto/tests/cases/plans/sources.rs | 143 +++++++++++----------
docs/source/library-user-guide/upgrading/56.0.0.md | 14 ++
3 files changed, 97 insertions(+), 66 deletions(-)
diff --git a/datafusion/datasource-csv/src/source.rs b/datafusion/datasource-csv/src/source.rs
index 0bfa499d7a..b62589a3e8 100644
--- a/datafusion/datasource-csv/src/source.rs
+++ b/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]),
};
diff --git a/datafusion/proto/tests/cases/plans/sources.rs b/datafusion/proto/tests/cases/plans/sources.rs
index 04d002e1e6..a215a01e84 100644
--- a/datafusion/proto/tests/cases/plans/sources.rs
+++ b/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(())
}
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 11a63a5aa5..4bc75d462d 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/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.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org