Decodes records from `input` returning the number of records and bytes read Note: this expects to be called with an empty `input` to signal EOF
(&mut self, input: &[u8], to_read: usize)
| 84 | /// |
| 85 | /// Note: this expects to be called with an empty `input` to signal EOF |
| 86 | pub fn decode(&mut self, input: &[u8], to_read: usize) -> Result<(usize, usize), ArrowError> { |
| 87 | if to_read == 0 { |
| 88 | return Ok((0, 0)); |
| 89 | } |
| 90 | |
| 91 | // Reserve sufficient capacity in offsets |
| 92 | self.offsets |
| 93 | .resize(self.offsets_len + to_read * self.num_columns, 0); |
| 94 | |
| 95 | // The current offset into `input` |
| 96 | let mut input_offset = 0; |
| 97 | |
| 98 | // The number of rows decoded in this pass |
| 99 | let mut read = 0; |
| 100 | |
| 101 | loop { |
| 102 | // Reserve necessary space in output data based on best estimate |
| 103 | let remaining_rows = to_read - read; |
| 104 | let capacity = remaining_rows * self.num_columns * AVERAGE_FIELD_SIZE; |
| 105 | let estimated_data = capacity.max(MIN_CAPACITY); |
| 106 | self.data.resize(self.data_len + estimated_data, 0); |
| 107 | |
| 108 | // Try to read a record |
| 109 | loop { |
| 110 | let (result, bytes_read, bytes_written, end_positions) = |
| 111 | self.delimiter.read_record( |
| 112 | &input[input_offset..], |
| 113 | &mut self.data[self.data_len..], |
| 114 | &mut self.offsets[self.offsets_len..], |
| 115 | ); |
| 116 | |
| 117 | self.current_field += end_positions; |
| 118 | self.offsets_len += end_positions; |
| 119 | input_offset += bytes_read; |
| 120 | self.data_len += bytes_written; |
| 121 | |
| 122 | match result { |
| 123 | ReadRecordResult::End | ReadRecordResult::InputEmpty => { |
| 124 | // Reached end of input |
| 125 | return Ok((read, input_offset)); |
| 126 | } |
| 127 | // Need to allocate more capacity |
| 128 | ReadRecordResult::OutputFull => break, |
| 129 | ReadRecordResult::OutputEndsFull => { |
| 130 | return Err(ArrowError::CsvError(format!( |
| 131 | "incorrect number of fields for line {}, expected {} got more than {}", |
| 132 | self.line_number, self.num_columns, self.current_field |
| 133 | ))); |
| 134 | } |
| 135 | ReadRecordResult::Record => { |
| 136 | if self.current_field != self.num_columns { |
| 137 | if self.truncated_rows && self.current_field < self.num_columns { |
| 138 | // If the number of fields is less than expected, pad with nulls |
| 139 | let fill_count = self.num_columns - self.current_field; |
| 140 | let fill_value = self.offsets[self.offsets_len - 1]; |
| 141 | self.offsets[self.offsets_len..self.offsets_len + fill_count] |
| 142 | .fill(fill_value); |
| 143 | self.offsets_len += fill_count; |