From 9d89eafaf3549e88f10d20e1adc1e8f1a25f7c6e Mon Sep 17 00:00:00 2001 From: osipovartem Date: Sun, 11 Oct 2026 11:18:44 +0300 Subject: [PATCH 1/3] Replace invalid UTF-8 in opt-in CSV decoded fields --- arrow-csv/src/reader/mod.rs | 30 ++++++ arrow-csv/src/reader/records.rs | 163 +++++++++++++++++++++++++++++--- 2 files changed, 178 insertions(+), 15 deletions(-) diff --git a/arrow-csv/src/reader/mod.rs b/arrow-csv/src/reader/mod.rs index 549d87c3b058..6be58b4b5d1e 100644 --- a/arrow-csv/src/reader/mod.rs +++ b/arrow-csv/src/reader/mod.rs @@ -1643,6 +1643,7 @@ pub struct ReaderBuilder { projection: Option>, /// Optional handler for records whose field count differs from the schema. record_error_handler: Option>, + replace_invalid_utf8: bool, numeric_boolean_values: bool, } @@ -1677,6 +1678,7 @@ impl ReaderBuilder { bounds: None, projection: None, record_error_handler: None, + replace_invalid_utf8: false, numeric_boolean_values: false, } } @@ -1796,6 +1798,13 @@ impl ReaderBuilder { self } + /// Replace malformed UTF-8 in decoded CSV fields with U+FFFD. + /// Disabled by default; only malformed records allocate replacement data. + pub fn with_replace_invalid_utf8(mut self, replace: bool) -> Self { + self.replace_invalid_utf8 = replace; + self + } + /// Create a new `Reader` from a non-buffered reader /// /// If `R: BufRead` consider using [`Self::build_buffered`] to avoid unnecessary additional @@ -1821,6 +1830,7 @@ impl ReaderBuilder { self.format.truncated_rows, ) .with_record_error_handler(self.record_error_handler) + .with_replace_invalid_utf8(self.replace_invalid_utf8) .with_ignored_prefix_options(self.format.comment, self.format.terminator) .with_utf8_preserving_syntax( self.format.delimiter.is_none_or(|byte| byte.is_ascii()) @@ -1961,6 +1971,26 @@ mod tests { assert_eq!(batches[1].column(1).as_string::().value(0), "after"); } + #[test] + fn test_replace_invalid_utf8_through_reader_builder() { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("msg", DataType::Utf8, false), + ])); + let csv = b"1,ok\n2,\xff\n3,after\n"; + let batches = ReaderBuilder::new(schema) + .with_batch_size(1) + .with_replace_invalid_utf8(true) + .build(Cursor::new(csv)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(batches.len(), 3); + assert_eq!(batches[0].column(1).as_string::().value(0), "ok"); + assert_eq!(batches[1].column(1).as_string::().value(0), "�"); + assert_eq!(batches[2].column(1).as_string::().value(0), "after"); + } + #[test] #[ignore = "local small-batch CSV recovery throughput benchmark"] fn bench_valid_unicode_csv_with_single_row_batches() { diff --git a/arrow-csv/src/reader/records.rs b/arrow-csv/src/reader/records.rs index 23b24f58c1a8..d8833ee932e6 100644 --- a/arrow-csv/src/reader/records.rs +++ b/arrow-csv/src/reader/records.rs @@ -21,6 +21,18 @@ use std::sync::Arc; use super::{CsvInvalidUtf8Error, CsvRecord, CsvRecordError, CsvRecordErrorHandler, NullRegex}; +#[derive(Debug)] +struct RejectMalformedRecord; + +impl CsvRecordErrorHandler for RejectMalformedRecord { + fn handle(&self, error: &CsvRecordError<'_>) -> Result<(), ArrowError> { + Err(ArrowError::CsvError(format!( + "incorrect number of fields for line {}, expected {} got {}", + error.line_number, error.expected_fields, error.actual_fields + ))) + } +} + #[derive(Debug)] struct QuotedEmptyFields { delimiter: u8, @@ -257,6 +269,9 @@ pub struct RecordDecoder { /// Optional recovery hook. Keeping this `None` selects the allocation-free strict path. record_error_handler: Option>, + /// Opt-in replacement of malformed decoded UTF-8 fields. + replace_invalid_utf8: bool, + /// One-based physical source line, kept separate from the record ordinal. source_lines: SourceLineCounter, @@ -299,6 +314,7 @@ impl RecordDecoder { quoted_empty: None, truncated_rows, record_error_handler: None, + replace_invalid_utf8: false, source_lines: SourceLineCounter { lf_count: 0, count_lf_independently: true, @@ -326,6 +342,11 @@ impl RecordDecoder { self } + pub fn with_replace_invalid_utf8(mut self, replace: bool) -> Self { + self.replace_invalid_utf8 = replace; + self + } + pub(super) fn with_ignored_prefix_options( mut self, comment: Option, @@ -369,6 +390,9 @@ impl RecordDecoder { Some(handler) => { self.decode_with_record_error_handler(input, to_read, handler.as_ref()) } + None if self.replace_invalid_utf8 => { + self.decode_with_record_error_handler(input, to_read, &RejectMalformedRecord) + } None => self.decode_strict(input, to_read), } } @@ -690,21 +714,26 @@ impl RecordDecoder { }) .flatten(); if let Some(field_index) = invalid_field { - source_lines.advance(&input[counted_until..input_offset]); - counted_until = input_offset; - handler.handle_invalid_utf8(&CsvInvalidUtf8Error { - line_number: self.line_number, - source_line_number: source_lines.record_start_line( - usize::try_from(self.delimiter.line()).unwrap_or(usize::MAX), - &self.record_bytes, - ), - byte_offset: self.record_byte_offset, - field_index, - record: &self.record_bytes, - })?; - self.skip_current_record(); - self.record_needs_utf8_validation = input_requires_utf8_validation; - continue; + if self.replace_invalid_utf8 { + self.replace_invalid_utf8_fields()?; + } else { + source_lines.advance(&input[counted_until..input_offset]); + counted_until = input_offset; + handler.handle_invalid_utf8(&CsvInvalidUtf8Error { + line_number: self.line_number, + source_line_number: source_lines.record_start_line( + usize::try_from(self.delimiter.line()) + .unwrap_or(usize::MAX), + &self.record_bytes, + ), + byte_offset: self.record_byte_offset, + field_index, + record: &self.record_bytes, + })?; + self.skip_current_record(); + self.record_needs_utf8_validation = input_requires_utf8_validation; + continue; + } } handler.handle_record(&CsvRecord { @@ -778,6 +807,24 @@ impl RecordDecoder { self.record_byte_offset = self.stream_offset; } + fn replace_invalid_utf8_fields(&mut self) -> Result<(), ArrowError> { + let row = self.data[self.record_data_start..self.data_len].to_vec(); + self.data.truncate(self.record_data_start); + let mut field_start = 0; + for end in &mut self.offsets[self.record_offsets_start..self.offsets_len] { + let field_end = *end; + let field = row.get(field_start..field_end).ok_or_else(|| { + ArrowError::CsvError("CSV decoded field offset is invalid".to_owned()) + })?; + self.data + .extend_from_slice(String::from_utf8_lossy(field).as_bytes()); + *end = self.data.len() - self.record_data_start; + field_start = field_end; + } + self.data_len = self.data.len(); + Ok(()) + } + /// Returns the current number of buffered records pub fn len(&self) -> usize { self.num_rows @@ -1410,6 +1457,92 @@ mod tests { assert_eq!(errors[0].record, b"2,\"before\n\xff\"\n"); } + #[test] + fn test_replace_invalid_utf8_preserves_records_and_field_offsets() { + let csv = b"1,ok\n2,\xff\xfe,extra\n3,after\n"; + let mut decoder = + RecordDecoder::new(Reader::new(), 3, true).with_replace_invalid_utf8(true); + let (rows, bytes) = decoder.decode(csv, 3).unwrap(); + assert_eq!((rows, bytes), (3, csv.len())); + let records = decoder.flush().unwrap(); + let values = records + .iter() + .map(|record| (record.get(0), record.get(1), record.get(2))) + .collect::>(); + assert_eq!( + values, + [("1", "ok", ""), ("2", "��", "extra"), ("3", "after", "")] + ); + } + + #[test] + fn test_replace_invalid_utf8_observes_repaired_record_without_error() { + let csv = b"1,\"bef\xffore\"\n2,after\n"; + let handler = Arc::new(CollectRecords::default()); + let mut decoder = RecordDecoder::new(Reader::new(), 2, false) + .with_replace_invalid_utf8(true) + .with_record_error_handler(Some(handler.clone())); + let (rows, bytes) = decoder.decode(csv, 2).unwrap(); + assert_eq!((rows, bytes), (2, csv.len())); + let records = decoder.flush().unwrap(); + assert_eq!( + records + .iter() + .map(|record| record.get(1)) + .collect::>(), + ["bef�ore", "after"] + ); + assert!(handler.invalid_utf8.lock().unwrap().is_empty()); + assert_eq!(handler.records.lock().unwrap().len(), 2); + } + + #[test] + fn test_replace_invalid_utf8_does_not_replace_valid_split_unicode() { + let csv = "1,é\n".as_bytes(); + let mut decoder = + RecordDecoder::new(Reader::new(), 2, false).with_replace_invalid_utf8(true); + let (first_rows, first_bytes) = decoder.decode(&csv[..3], 1).unwrap(); + let (later_rows, later_bytes) = decoder.decode(&csv[3..], 1).unwrap(); + assert_eq!(first_rows + later_rows, 1); + assert_eq!(first_bytes + later_bytes, csv.len()); + let records = decoder.flush().unwrap(); + assert_eq!(records.iter().next().unwrap().get(1), "é"); + } + + #[test] + fn test_replace_invalid_utf8_still_rejects_field_count_mismatch() { + let mut decoder = + RecordDecoder::new(Reader::new(), 2, false).with_replace_invalid_utf8(true); + let error = decoder.decode(b"1,ok,extra\n", 1).unwrap_err(); + assert!(error.to_string().contains("incorrect number of fields")); + } + + #[test] + fn test_replace_invalid_utf8_split_across_field_boundary() { + let mut decoder = + RecordDecoder::new(Reader::new(), 3, false).with_replace_invalid_utf8(true); + let csv = b"1,\xc3,\xa9\n"; + assert_eq!(decoder.decode(csv, 1).unwrap(), (1, csv.len())); + let records = decoder.flush().unwrap(); + let record = records.iter().next().unwrap(); + assert_eq!( + (record.get(0), record.get(1), record.get(2)), + ("1", "�", "�") + ); + } + + #[test] + fn test_replace_invalid_utf8_after_high_byte_quote() { + let mut builder = csv_core::ReaderBuilder::new(); + builder.quote(b'\xc3'); + let mut decoder = + RecordDecoder::new(builder.build(), 2, false).with_replace_invalid_utf8(true); + let csv = b"1,\xc3\xa9\xc3\xa9\n"; + assert_eq!(decoder.decode(csv, 1).unwrap(), (1, csv.len())); + let records = decoder.flush().unwrap(); + assert_eq!(records.iter().next().unwrap().get(1), "��"); + } + #[test] fn test_validated_header_preserves_later_recovery_offsets() { let csv = b"id,msg\n2,\xff\n"; From e45e8c1577173c279244a3b73221c9cf032c860b Mon Sep 17 00:00:00 2001 From: osipovartem Date: Sun, 11 Oct 2026 11:22:21 +0300 Subject: [PATCH 2/3] Keep CSV UTF-8 replacement on strict fast path without handlers --- arrow-csv/src/reader/mod.rs | 25 ++++++++++++++++ arrow-csv/src/reader/records.rs | 51 ++++++++++++++++++++++----------- 2 files changed, 59 insertions(+), 17 deletions(-) diff --git a/arrow-csv/src/reader/mod.rs b/arrow-csv/src/reader/mod.rs index 6be58b4b5d1e..fe8aecfa3eb3 100644 --- a/arrow-csv/src/reader/mod.rs +++ b/arrow-csv/src/reader/mod.rs @@ -1991,6 +1991,31 @@ mod tests { assert_eq!(batches[2].column(1).as_string::().value(0), "after"); } + #[test] + #[ignore = "focused CSV UTF-8 replacement throughput benchmark"] + fn bench_valid_unicode_csv_with_optional_replacement() { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("msg", DataType::Utf8, false), + ])); + let csv = "1,é\n".repeat(1_000_000); + for replace in [false, true] { + let start = std::time::Instant::now(); + let rows = ReaderBuilder::new(Arc::clone(&schema)) + .with_batch_size(4096) + .with_replace_invalid_utf8(replace) + .build(Cursor::new(csv.as_bytes())) + .unwrap() + .map(|batch| batch.unwrap().num_rows()) + .sum::(); + assert_eq!(rows, 1_000_000); + eprintln!( + "replace_invalid_utf8={replace} elapsed={:?}", + start.elapsed() + ); + } + } + #[test] #[ignore = "local small-batch CSV recovery throughput benchmark"] fn bench_valid_unicode_csv_with_single_row_batches() { diff --git a/arrow-csv/src/reader/records.rs b/arrow-csv/src/reader/records.rs index d8833ee932e6..3b25a14a9d1d 100644 --- a/arrow-csv/src/reader/records.rs +++ b/arrow-csv/src/reader/records.rs @@ -21,18 +21,6 @@ use std::sync::Arc; use super::{CsvInvalidUtf8Error, CsvRecord, CsvRecordError, CsvRecordErrorHandler, NullRegex}; -#[derive(Debug)] -struct RejectMalformedRecord; - -impl CsvRecordErrorHandler for RejectMalformedRecord { - fn handle(&self, error: &CsvRecordError<'_>) -> Result<(), ArrowError> { - Err(ArrowError::CsvError(format!( - "incorrect number of fields for line {}, expected {} got {}", - error.line_number, error.expected_fields, error.actual_fields - ))) - } -} - #[derive(Debug)] struct QuotedEmptyFields { delimiter: u8, @@ -390,9 +378,6 @@ impl RecordDecoder { Some(handler) => { self.decode_with_record_error_handler(input, to_read, handler.as_ref()) } - None if self.replace_invalid_utf8 => { - self.decode_with_record_error_handler(input, to_read, &RejectMalformedRecord) - } None => self.decode_strict(input, to_read), } } @@ -825,6 +810,24 @@ impl RecordDecoder { Ok(()) } + fn replace_invalid_utf8_batch(&mut self) -> Result<(), ArrowError> { + let original = self.data[..self.data_len].to_vec(); + self.data.clear(); + let mut field_start = 0; + for end in &mut self.offsets[1..self.offsets_len] { + let field_end = *end; + let field = original.get(field_start..field_end).ok_or_else(|| { + ArrowError::CsvError("CSV decoded field offset is invalid".to_owned()) + })?; + self.data + .extend_from_slice(String::from_utf8_lossy(field).as_bytes()); + *end = self.data.len(); + field_start = field_end; + } + self.data_len = self.data.len(); + Ok(()) + } + /// Returns the current number of buffered records pub fn len(&self) -> usize { self.num_rows @@ -871,6 +874,19 @@ impl RecordDecoder { }) })?; + if self.replace_invalid_utf8 && self.record_error_handler.is_none() { + let decoded = std::str::from_utf8(&self.data[..self.data_len]); + let invalid_field = match decoded { + Ok(data) => self.offsets[1..self.offsets_len] + .iter() + .any(|end| !data.is_char_boundary(*end)), + Err(_) => true, + }; + if invalid_field { + self.replace_invalid_utf8_batch()?; + } + } + // Need to truncate data t1o the actual amount of data read let data = std::str::from_utf8(&self.data[..self.data_len]).map_err(|e| { let valid_up_to = e.valid_up_to(); @@ -1519,8 +1535,9 @@ mod tests { #[test] fn test_replace_invalid_utf8_split_across_field_boundary() { - let mut decoder = - RecordDecoder::new(Reader::new(), 3, false).with_replace_invalid_utf8(true); + let mut decoder = RecordDecoder::new(Reader::new(), 3, false) + .with_utf8_preserving_syntax(true) + .with_replace_invalid_utf8(true); let csv = b"1,\xc3,\xa9\n"; assert_eq!(decoder.decode(csv, 1).unwrap(), (1, csv.len())); let records = decoder.flush().unwrap(); From 95af4707b289c8dfdbbf6a4a3bb7647da6865bb1 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Sun, 11 Oct 2026 11:32:57 +0300 Subject: [PATCH 3/3] Apply CSV UTF-8 replacement during schema inference --- arrow-csv/src/reader/mod.rs | 155 ++++++++++++++++++++++++++++++------ 1 file changed, 131 insertions(+), 24 deletions(-) diff --git a/arrow-csv/src/reader/mod.rs b/arrow-csv/src/reader/mod.rs index fe8aecfa3eb3..91a57345243f 100644 --- a/arrow-csv/src/reader/mod.rs +++ b/arrow-csv/src/reader/mod.rs @@ -510,6 +510,7 @@ pub struct Format { null_regex: NullRegex, truncated_rows: bool, preserve_quoted_empty: bool, + replace_invalid_utf8: bool, } impl Format { @@ -585,6 +586,13 @@ impl Format { self } + /// Replace malformed UTF-8 in decoded CSV fields during schema inference. + /// Disabled by default. + pub fn with_replace_invalid_utf8(mut self, replace: bool) -> Self { + self.replace_invalid_utf8 = replace; + self + } + /// Whether to allow truncated rows when parsing. /// /// By default this is set to `false` and will error if the CSV rows have different lengths. @@ -648,7 +656,21 @@ impl Format { // get or create header names // when has_header is false, creates default column names with column_ prefix - let headers: Vec = if self.header { + let headers: Vec = if self.replace_invalid_utf8 { + csv_reader + .byte_headers() + .map_err(map_csv_error)? + .iter() + .enumerate() + .map(|(index, bytes)| { + if self.header { + String::from_utf8_lossy(bytes).into_owned() + } else { + format!("column_{}", index + 1) + } + }) + .collect() + } else if self.header { let headers = &csv_reader.headers().map_err(map_csv_error)?.clone(); headers.iter().map(|s| s.to_string()).collect() } else { @@ -664,25 +686,48 @@ impl Format { let mut records_count = 0; - let mut record = StringRecord::new(); let max_records = max_records.unwrap_or(usize::MAX); - while records_count < max_records { - if !csv_reader.read_record(&mut record).map_err(map_csv_error)? { - break; + if self.replace_invalid_utf8 { + let mut record = csv::ByteRecord::new(); + while records_count < max_records + && csv_reader + .read_byte_record(&mut record) + .map_err(map_csv_error)? + { + records_count += 1; + for (i, column_type) in column_types.iter_mut().enumerate().take(header_length) { + if let Some(bytes) = record.get(i) { + let string = String::from_utf8_lossy(bytes); + if !self.null_regex.is_null(&string) { + if SNOWFLAKE { + column_type.update_snowflake(&string); + } else if DECIMAL { + column_type.update_decimal::(&string); + } else { + column_type.update(&string); + } + } + } + } } - records_count += 1; - - // Note since we may be looking at a sample of the data, we make the safe assumption that - // they could be nullable - for (i, column_type) in column_types.iter_mut().enumerate().take(header_length) { - if let Some(string) = record.get(i) { - if !self.null_regex.is_null(string) { - if SNOWFLAKE { - column_type.update_snowflake(string); - } else if DECIMAL { - column_type.update_decimal::(string); - } else { - column_type.update(string); + } else { + let mut record = StringRecord::new(); + while records_count < max_records { + if !csv_reader.read_record(&mut record).map_err(map_csv_error)? { + break; + } + records_count += 1; + // Inferred columns remain nullable because this is only a sample. + for (i, column_type) in column_types.iter_mut().enumerate().take(header_length) { + if let Some(string) = record.get(i) { + if !self.null_regex.is_null(string) { + if SNOWFLAKE { + column_type.update_snowflake(string); + } else if DECIMAL { + column_type.update_decimal::(string); + } else { + column_type.update(string); + } } } } @@ -715,7 +760,21 @@ impl Format { prefix: Vec::new(), }; let mut header_reader = self.build_reader(recording); - let headers: Vec = if self.header { + let headers: Vec = if self.replace_invalid_utf8 { + header_reader + .byte_headers() + .map_err(map_csv_error)? + .iter() + .enumerate() + .map(|(index, bytes)| { + if self.header { + String::from_utf8_lossy(bytes).into_owned() + } else { + format!("column_{}", index + 1) + } + }) + .collect() + } else if self.header { header_reader .headers() .map_err(map_csv_error)? @@ -1643,7 +1702,6 @@ pub struct ReaderBuilder { projection: Option>, /// Optional handler for records whose field count differs from the schema. record_error_handler: Option>, - replace_invalid_utf8: bool, numeric_boolean_values: bool, } @@ -1678,7 +1736,6 @@ impl ReaderBuilder { bounds: None, projection: None, record_error_handler: None, - replace_invalid_utf8: false, numeric_boolean_values: false, } } @@ -1799,9 +1856,9 @@ impl ReaderBuilder { } /// Replace malformed UTF-8 in decoded CSV fields with U+FFFD. - /// Disabled by default; only malformed records allocate replacement data. + /// Disabled by default; malformed batches may allocate replacement data. pub fn with_replace_invalid_utf8(mut self, replace: bool) -> Self { - self.replace_invalid_utf8 = replace; + self.format.replace_invalid_utf8 = replace; self } @@ -1830,7 +1887,7 @@ impl ReaderBuilder { self.format.truncated_rows, ) .with_record_error_handler(self.record_error_handler) - .with_replace_invalid_utf8(self.replace_invalid_utf8) + .with_replace_invalid_utf8(self.format.replace_invalid_utf8) .with_ignored_prefix_options(self.format.comment, self.format.terminator) .with_utf8_preserving_syntax( self.format.delimiter.is_none_or(|byte| byte.is_ascii()) @@ -2016,6 +2073,56 @@ mod tests { } } + #[test] + fn test_infer_schema_replaces_invalid_utf8_in_header_and_data() { + let csv = b"id,m\xffg\n1,\xfe\n"; + assert!( + Format::default() + .with_header(true) + .infer_schema_with_snowflake_types(Cursor::new(csv), None) + .is_err() + ); + for preserve_quoted_empty in [false, true] { + let (schema, _, records) = Format::default() + .with_header(true) + .with_preserve_quoted_empty(preserve_quoted_empty) + .with_replace_invalid_utf8(true) + .infer_schema_with_snowflake_types(Cursor::new(csv), None) + .unwrap(); + assert_eq!(records, 1); + assert_eq!(schema.field(1).name(), "m�g"); + assert_eq!(schema.field(1).data_type(), &DataType::Utf8); + } + } + + #[test] + #[ignore = "focused sparse-error CSV replacement benchmark"] + fn bench_sparse_invalid_utf8_replacement() { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("msg", DataType::Utf8, false), + ])); + let mut csv = b"1,ok\n".repeat(1_000_000); + for invalid in [false, true] { + if invalid { + csv[500_000 * 5 + 2] = 0xff; + } + let start = std::time::Instant::now(); + let rows = ReaderBuilder::new(Arc::clone(&schema)) + .with_batch_size(4096) + .with_replace_invalid_utf8(true) + .build(Cursor::new(&csv)) + .unwrap() + .map(|batch| batch.unwrap().num_rows()) + .sum::(); + assert_eq!(rows, 1_000_000); + eprintln!( + "sparse_invalid_utf8={invalid} elapsed={:?}", + start.elapsed() + ); + } + } + #[test] #[ignore = "local small-batch CSV recovery throughput benchmark"] fn bench_valid_unicode_csv_with_single_row_batches() {