This is an automated email from the ASF dual-hosted git repository.
Jefffrey pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new f445af2d1f Report the number of rows padded by `with_truncated_rows`
(#10579)
f445af2d1f is described below
commit f445af2d1f340adf81a465c9a21f649cd598f2ca
Author: Andrea Bozzo <[email protected]>
AuthorDate: Mon Aug 10 14:20:58 2026 +0200
Report the number of rows padded by `with_truncated_rows` (#10579)
# Which issue does this PR close?
Closes #10577.
# Rationale for this change
`with_truncated_rows(true)` repairs a row that has fewer fields than the
schema by
padding it, and reports nothing about having done so. For a consumer
that reports on
data quality, a repaired parse and a clean parse are different outcomes,
and today
they are indistinguishable.
The count cannot be recovered after decoding. Padding fills offsets to
produce
zero-length fields, and `NullRegex` later turns those into nulls, so a
padded field
is byte-identical to a genuinely empty trailing field:
```
name,age,city
Alice,25,Rome -> ["Alice", "25", "Rome"]
Carol,35, -> ["Carol", "35", NULL] three fields, the last one empty
Bob,30 -> ["Bob", "30", NULL] two fields, padded
```
Disabling the null regex does not help, both cases become `""` and stay
identical.
Nothing on `ReaderBuilder`, `Format`, `Reader`, `BufReader` or `Decoder`
exposes the
information either.
This came out of dataprof, where the Arrow-backed CSV engine was the
only path that
reported a file of short rows with a perfect consistency score. The
workaround
shipped there is a pre-scan with the `csv` crate purely to recover the
count, which
costs a second full read of the file. On a 218 MB, 2M-row file that
measured at
roughly 3% of profiling wall time, because per-value analysis dominates
the parse.
So the workaround is viable, but it is a whole extra pass to recover a
number the
decoder already had and threw away.
# What changes are included in this PR?
A single counter, threaded up to the public types.
`arrow-csv/src/reader/records.rs`:
* `RecordDecoder` gains a `truncated_row_count` field, incremented in
the one branch
of `decode` that pads a short row.
* `RecordDecoder::truncated_row_count()` returns it.
* `flush` deliberately leaves the counter alone, so it accumulates
across batches.
`clear` resets it, because `clear` discards the buffered rows the count
refers to.
That is what keeps skipped rows out of the total, see below.
`arrow-csv/src/reader/mod.rs`:
* `Decoder::truncated_row_count()` and
`BufReader::truncated_row_count()` forward it.
`Reader<R>` is an alias for `BufReader<StdBufReader<R>>`, so the
accessor covers
both.
Two semantics worth calling out, both documented on the accessors:
* **The count is cumulative, not per batch.** `RecordDecoder` is reused
across
`flush` calls and the counter survives them. Reading it between batches
gives a
running total of the rows decoded so far, reading it after the input is
exhausted
gives the total for the whole input. It is meaningful at either point as
long as
that is understood, so it is stated rather than restricted.
* **Skipped rows do not contribute.** The header row and any rows before
the start
bound go through `RecordDecoder::decode` too, and would otherwise be
counted if
they were short. They are discarded via `clear`, which now resets the
counter with
them, so the total only ever covers rows that reached a batch.
# Are these changes tested?
Yes, at both levels, and each test was confirmed to fail against
unpatched code.
`records.rs`:
* `test_truncated_rows` extended to assert the count.
* `test_truncated_row_count_not_reset_by_flush`.
* `test_truncated_row_count_reset_by_clear`.
`mod.rs`:
* `test_truncated_row_count_counts_padded_rows`, one short row, count is
1.
* `test_truncated_row_count_ignores_empty_trailing_field`, a row with a
genuinely
empty trailing field produces the same null as a padded row but the
count stays 0.
This is the case that proves the count is not inferred from nulls.
* `test_truncated_row_count_clean_file`, count is 0.
* `test_truncated_row_count_without_truncated_rows`, the short row
errors as before
and the accessor still returns 0.
* `test_truncated_row_count_accumulates_across_batches`, `batch_size` 2
over 6 short
rows, asserting the running total is 2, 4, 6 rather than 2 each time.
* `test_truncated_row_count_excludes_skipped_rows`, a header shorter
than the schema
is padded while being skipped and does not count.
* `test_truncated_row_count_on_decoder`, the push-based `Decoder` path.
The two parts of the change, the increment and the `clear` reset, were
reverted
separately to confirm each is independently covered.
`cargo test -p arrow-csv`, `cargo fmt --all` and
`cargo clippy -p arrow-csv --all-targets -- -D warnings` are clean.
# Are there any user-facing changes?
Yes, one additive accessor, `truncated_row_count()`, on `BufReader` (and
therefore
`Reader`) and on `Decoder`. There are no breaking changes, no behaviour
changes to
parsing, and no new configuration.
Co-authored-by: Jeffrey Vo <[email protected]>
---
arrow-csv/src/reader/mod.rs | 194 ++++++++++++++++++++++++++++++++++++++++
arrow-csv/src/reader/records.rs | 55 ++++++++++++
2 files changed, 249 insertions(+)
diff --git a/arrow-csv/src/reader/mod.rs b/arrow-csv/src/reader/mod.rs
index a286f19f7e..9007acc14e 100644
--- a/arrow-csv/src/reader/mod.rs
+++ b/arrow-csv/src/reader/mod.rs
@@ -519,6 +519,46 @@ where
}
}
+impl<R> BufReader<R> {
+ /// The number of rows padded because they had fewer fields than the schema
+ ///
+ /// Always 0 unless [`ReaderBuilder::with_truncated_rows`] was set to
`true`.
+ ///
+ /// The count is cumulative over the lifetime of this reader, so reading it
+ /// between batches yields a running total of the rows read so far, and
reading it
+ /// once the reader is exhausted yields the total for the whole input.
Rows that
+ /// are skipped rather than read into a batch, such as a header row or
rows before
+ /// the start bound, do not contribute.
+ ///
+ /// A padded row is indistinguishable from a row with genuinely empty
trailing
+ /// fields once it has been read, so this counter is the only way to tell
the two
+ /// apart.
+ ///
+ /// ```
+ /// # use std::io::Cursor;
+ /// # use std::sync::Arc;
+ /// # use arrow_csv::ReaderBuilder;
+ /// # use arrow_schema::{DataType, Field, Schema};
+ /// #
+ /// let schema = Arc::new(Schema::new(vec![
+ /// Field::new("a", DataType::Int32, true),
+ /// Field::new("b", DataType::Int32, true),
+ /// ]));
+ ///
+ /// let mut reader = ReaderBuilder::new(schema)
+ /// .with_truncated_rows(true)
+ /// .build(Cursor::new("1,2\n3\n"))
+ /// .unwrap();
+ ///
+ /// let batches = reader.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
+ /// assert_eq!(batches[0].num_rows(), 2);
+ /// assert_eq!(reader.truncated_row_count(), 1);
+ /// ```
+ pub fn truncated_row_count(&self) -> usize {
+ self.decoder.truncated_row_count()
+ }
+}
+
impl<R: Read> Reader<R> {
/// Returns the schema of the reader, useful for getting the schema
without reading
/// record batches
@@ -704,6 +744,23 @@ impl Decoder {
pub fn capacity(&self) -> usize {
self.batch_size - self.record_decoder.len()
}
+
+ /// The number of rows padded because they had fewer fields than the schema
+ ///
+ /// Always 0 unless [`ReaderBuilder::with_truncated_rows`] was set to
`true`.
+ ///
+ /// The count is cumulative over the lifetime of this decoder and is not
reset by
+ /// [`Self::flush`], so reading it between batches yields a running total
of the
+ /// rows decoded so far, and reading it once the input is exhausted yields
the
+ /// total for the whole stream. Rows that are skipped rather than decoded
into a
+ /// batch, such as a header row or rows before the start bound, do not
contribute.
+ ///
+ /// A padded row is indistinguishable from a row with genuinely empty
trailing
+ /// fields once it has been decoded, so this counter is the only way to
tell the
+ /// two apart.
+ pub fn truncated_row_count(&self) -> usize {
+ self.record_decoder.truncated_row_count()
+ }
}
fn validate_header(rows: &StringRecords<'_>, fields: &Fields) -> Result<(),
ArrowError> {
@@ -2635,6 +2692,143 @@ mod tests {
assert!(dob.is_null(5));
}
+ /// Schema used by the `truncated_row_count` tests below
+ fn truncated_row_count_schema() -> SchemaRef {
+ Arc::new(Schema::new(vec![
+ Field::new("name", DataType::Utf8, true),
+ Field::new("age", DataType::Int32, true),
+ Field::new("city", DataType::Utf8, true),
+ ]))
+ }
+
+ #[test]
+ fn test_truncated_row_count_counts_padded_rows() {
+ let data = "name,age,city\nAlice,25,Rome\nBob,30\n";
+
+ let mut reader = ReaderBuilder::new(truncated_row_count_schema())
+ .with_header(true)
+ .with_truncated_rows(true)
+ .build(Cursor::new(data))
+ .unwrap();
+
+ let batches = reader.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
+ assert_eq!(batches[0].num_rows(), 2);
+ assert_eq!(reader.truncated_row_count(), 1);
+ }
+
+ #[test]
+ fn test_truncated_row_count_ignores_empty_trailing_field() {
+ // "Carol,35," has all three fields, the last one just happens to be
empty, so it
+ // parses to the same null as a padded row would. The count must not
be inferred
+ // from the nulls in the batch
+ let data = "name,age,city\nAlice,25,Rome\nCarol,35,\n";
+
+ let mut reader = ReaderBuilder::new(truncated_row_count_schema())
+ .with_header(true)
+ .with_truncated_rows(true)
+ .build(Cursor::new(data))
+ .unwrap();
+
+ let batches = reader.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
+ let batch = &batches[0];
+ assert_eq!(batch.num_rows(), 2);
+ assert!(batch.column(2).is_null(1));
+ assert_eq!(reader.truncated_row_count(), 0);
+ }
+
+ #[test]
+ fn test_truncated_row_count_clean_file() {
+ let data = "name,age,city\nAlice,25,Rome\nBob,30,Milan\n";
+
+ let mut reader = ReaderBuilder::new(truncated_row_count_schema())
+ .with_header(true)
+ .with_truncated_rows(true)
+ .build(Cursor::new(data))
+ .unwrap();
+
+ let batches = reader.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
+ assert_eq!(batches[0].num_rows(), 2);
+ assert_eq!(reader.truncated_row_count(), 0);
+ }
+
+ #[test]
+ fn test_truncated_row_count_without_truncated_rows() {
+ let data = "name,age,city\nAlice,25,Rome\nBob,30\n";
+
+ let mut reader = ReaderBuilder::new(truncated_row_count_schema())
+ .with_header(true)
+ .with_truncated_rows(false)
+ .build(Cursor::new(data))
+ .unwrap();
+
+ // The short row is an error rather than something to count
+ let err = reader.by_ref().collect::<Result<Vec<_>, _>>().unwrap_err();
+ assert!(
+ err.to_string().contains("incorrect number of fields"),
+ "{err}"
+ );
+ assert_eq!(reader.truncated_row_count(), 0);
+ }
+
+ #[test]
+ fn test_truncated_row_count_accumulates_across_batches() {
+ // Six short rows read two at a time
+ let data = "name,age,city\nn0,0\nn1,1\nn2,2\nn3,3\nn4,4\nn5,5\n";
+
+ let mut reader = ReaderBuilder::new(truncated_row_count_schema())
+ .with_header(true)
+ .with_truncated_rows(true)
+ .with_batch_size(2)
+ .build(Cursor::new(data))
+ .unwrap();
+
+ let mut running = vec![];
+ while let Some(batch) = reader.next().transpose().unwrap() {
+ assert_eq!(batch.num_rows(), 2);
+ running.push(reader.truncated_row_count());
+ }
+
+ // A running total, not a per batch count
+ assert_eq!(running, vec![2, 4, 6]);
+ assert_eq!(reader.truncated_row_count(), 6);
+ }
+
+ #[test]
+ fn test_truncated_row_count_excludes_skipped_rows() {
+ // The header is one field short of the schema, so skipping it pads
it. Skipped
+ // rows never reach a batch and must not be counted
+ let data = "name,age\nAlice,25,Rome\nBob,30,Milan\n";
+
+ let mut reader = ReaderBuilder::new(truncated_row_count_schema())
+ .with_header(true)
+ .with_truncated_rows(true)
+ .build(Cursor::new(data))
+ .unwrap();
+
+ let batches = reader.by_ref().collect::<Result<Vec<_>, _>>().unwrap();
+ assert_eq!(batches[0].num_rows(), 2);
+ assert_eq!(reader.truncated_row_count(), 0);
+ }
+
+ #[test]
+ fn test_truncated_row_count_on_decoder() {
+ let data = "1,2\n3\n";
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ ]));
+
+ let mut decoder = ReaderBuilder::new(schema)
+ .with_truncated_rows(true)
+ .build_decoder();
+
+ assert_eq!(decoder.truncated_row_count(), 0);
+ let decoded = decoder.decode(data.as_bytes()).unwrap();
+ assert_eq!(decoded, data.len());
+ decoder.flush().unwrap().unwrap();
+ assert_eq!(decoder.truncated_row_count(), 1);
+ }
+
#[test]
fn test_truncated_rows_not_nullable_error() {
let data = "a,b,c\n1,2,3\n4,5";
diff --git a/arrow-csv/src/reader/records.rs b/arrow-csv/src/reader/records.rs
index a6d54f867c..7f22455106 100644
--- a/arrow-csv/src/reader/records.rs
+++ b/arrow-csv/src/reader/records.rs
@@ -62,6 +62,13 @@ pub struct RecordDecoder {
/// Default value is false
/// When enabled fills in missing columns with null
truncated_rows: bool,
+
+ /// The number of rows padded because they had fewer fields than expected
+ ///
+ /// Only incremented when `truncated_rows` is enabled. Cumulative over the
+ /// lifetime of this decoder, i.e. not reset by [`Self::flush`], but reset
by
+ /// [`Self::clear`] along with the buffered rows it counted
+ truncated_row_count: usize,
}
impl RecordDecoder {
@@ -77,6 +84,7 @@ impl RecordDecoder {
data: vec![],
num_rows: 0,
truncated_rows,
+ truncated_row_count: 0,
}
}
@@ -141,6 +149,7 @@ impl RecordDecoder {
self.offsets[self.offsets_len..self.offsets_len + fill_count]
.fill(fill_value);
self.offsets_len += fill_count;
+ self.truncated_row_count += 1;
} else {
return Err(ArrowError::CsvError(format!(
"incorrect number of fields for line {},
expected {} got {}",
@@ -180,12 +189,22 @@ impl RecordDecoder {
self.num_rows == 0
}
+ /// Returns the number of rows padded because they had fewer fields than
expected
+ ///
+ /// Cumulative across [`Self::flush`] calls, reset by [`Self::clear`]
+ pub fn truncated_row_count(&self) -> usize {
+ self.truncated_row_count
+ }
+
/// Clears the current contents of the decoder
pub fn clear(&mut self) {
// This does not reset current_field to allow clearing part way
through a record
self.offsets_len = 1;
self.data_len = 0;
self.num_rows = 0;
+ // The rows counted so far are being discarded along with the buffered
data,
+ // so they must not be reported as padded
+ self.truncated_row_count = 0;
}
/// Flushes the current contents of the reader
@@ -237,6 +256,8 @@ impl RecordDecoder {
let num_rows = self.num_rows;
// Reset state
+ // `truncated_row_count` is deliberately left alone so that it
accumulates
+ // across the batches produced by a single decoder
self.offsets_len = 1;
self.data_len = 0;
self.num_rows = 0;
@@ -403,6 +424,40 @@ mod tests {
let (read, bytes) = decoder.decode(csv.as_bytes(), 5).unwrap();
assert_eq!(read, 5);
assert_eq!(bytes, csv.len());
+ // Only "v" is short, the rows starting with a delimiter have both
fields
+ assert_eq!(decoder.truncated_row_count(), 1);
+ }
+
+ #[test]
+ fn test_truncated_row_count_not_reset_by_flush() {
+ let csv = "a\nb,2\nc\n";
+ let mut decoder = RecordDecoder::new(Reader::new(), 2, true);
+
+ let (read, _) = decoder.decode(csv.as_bytes(), 2).unwrap();
+ assert_eq!(read, 2);
+ assert_eq!(decoder.truncated_row_count(), 1);
+ decoder.flush().unwrap();
+ assert_eq!(decoder.truncated_row_count(), 1);
+
+ let (read, _) = decoder.decode(&csv.as_bytes()[6..], 1).unwrap();
+ assert_eq!(read, 1);
+ assert_eq!(decoder.truncated_row_count(), 2);
+ decoder.flush().unwrap();
+ assert_eq!(decoder.truncated_row_count(), 2);
+ }
+
+ #[test]
+ fn test_truncated_row_count_reset_by_clear() {
+ let csv = "a\nb,2\n";
+ let mut decoder = RecordDecoder::new(Reader::new(), 2, true);
+
+ let (read, _) = decoder.decode(csv.as_bytes(), 2).unwrap();
+ assert_eq!(read, 2);
+ assert_eq!(decoder.truncated_row_count(), 1);
+
+ // The rows are discarded, so the padding done to them is discarded too
+ decoder.clear();
+ assert_eq!(decoder.truncated_row_count(), 0);
}
/// Regression test for an overflow path found by the `arrow-csv`
