Flushes the current contents of the reader
(&mut self)
| 190 | |
| 191 | /// Flushes the current contents of the reader |
| 192 | pub fn flush(&mut self) -> Result<StringRecords<'_>, ArrowError> { |
| 193 | if self.current_field != 0 { |
| 194 | return Err(ArrowError::CsvError( |
| 195 | "Cannot flush part way through record".to_string(), |
| 196 | )); |
| 197 | } |
| 198 | |
| 199 | // csv_core::Reader writes end offsets relative to the start of the row |
| 200 | // Therefore scan through and offset these based on the cumulative row offsets |
| 201 | let mut row_offset: usize = 0; |
| 202 | self.offsets[1..self.offsets_len] |
| 203 | .chunks_exact_mut(self.num_columns) |
| 204 | .try_for_each(|row| -> Result<(), ArrowError> { |
| 205 | let offset = row_offset; |
| 206 | row.iter_mut().try_for_each(|x| -> Result<(), ArrowError> { |
| 207 | *x = x.checked_add(offset).ok_or_else(|| { |
| 208 | ArrowError::CsvError( |
| 209 | "CSV record offsets overflowed usize while flushing".to_string(), |
| 210 | ) |
| 211 | })?; |
| 212 | row_offset = *x; |
| 213 | Ok(()) |
| 214 | }) |
| 215 | })?; |
| 216 | |
| 217 | // Need to truncate data t1o the actual amount of data read |
| 218 | let data = std::str::from_utf8(&self.data[..self.data_len]).map_err(|e| { |
| 219 | let valid_up_to = e.valid_up_to(); |
| 220 | |
| 221 | // We can't use binary search because of empty fields |
| 222 | let idx = self.offsets[..self.offsets_len] |
| 223 | .iter() |
| 224 | .rposition(|x| *x <= valid_up_to) |
| 225 | .unwrap(); |
| 226 | |
| 227 | let field = idx % self.num_columns + 1; |
| 228 | let line_offset = self.line_number - self.num_rows; |
| 229 | let line = line_offset + idx / self.num_columns; |
| 230 | |
| 231 | ArrowError::CsvError(format!( |
| 232 | "Encountered invalid UTF-8 data for line {line} and field {field}" |
| 233 | )) |
| 234 | })?; |
| 235 | |
| 236 | let offsets = &self.offsets[..self.offsets_len]; |
| 237 | let num_rows = self.num_rows; |
| 238 | |
| 239 | // Reset state |
| 240 | self.offsets_len = 1; |
| 241 | self.data_len = 0; |
| 242 | self.num_rows = 0; |
| 243 | |
| 244 | Ok(StringRecords { |
| 245 | num_rows, |
| 246 | num_columns: self.num_columns, |
| 247 | offsets, |
| 248 | data, |
| 249 | }) |