Reads the next `RecordBatch` from the Avro file, or `Ok(None)` on EOF. Batches are bounded by `batch_size`; a single OCF block may yield multiple batches, and a batch may also span multiple blocks.
(&mut self)
| 1358 | /// Batches are bounded by `batch_size`; a single OCF block may yield multiple batches, |
| 1359 | /// and a batch may also span multiple blocks. |
| 1360 | fn read(&mut self) -> Result<Option<RecordBatch>, AvroError> { |
| 1361 | 'outer: while !self.finished && !self.decoder.batch_is_full() { |
| 1362 | while self.block_cursor == self.block_data.len() { |
| 1363 | let buf = self.reader.fill_buf()?; |
| 1364 | if buf.is_empty() { |
| 1365 | self.finished = true; |
| 1366 | break 'outer; |
| 1367 | } |
| 1368 | // Try to decode another block from the buffered reader. |
| 1369 | let consumed = self.block_decoder.decode(buf)?; |
| 1370 | self.reader.consume(consumed); |
| 1371 | if let Some(block) = self.block_decoder.flush() { |
| 1372 | // Successfully decoded a block. |
| 1373 | self.block_data = if let Some(ref codec) = self.header.compression()? { |
| 1374 | let decompressed: Vec<u8> = codec.decompress(&block.data)?; |
| 1375 | decompressed |
| 1376 | } else { |
| 1377 | block.data |
| 1378 | }; |
| 1379 | self.block_count = block.count; |
| 1380 | self.block_cursor = 0; |
| 1381 | } else if consumed == 0 { |
| 1382 | // The block decoder made no progress on a non-empty buffer. |
| 1383 | return Err(AvroError::ParseError( |
| 1384 | "Could not decode next Avro block from partial data".to_string(), |
| 1385 | )); |
| 1386 | } |
| 1387 | } |
| 1388 | // Decode as many rows as will fit in the current batch |
| 1389 | if self.block_cursor < self.block_data.len() { |
| 1390 | let (consumed, records_decoded) = self |
| 1391 | .decoder |
| 1392 | .decode_block(&self.block_data[self.block_cursor..], self.block_count)?; |
| 1393 | self.block_cursor += consumed; |
| 1394 | self.block_count -= records_decoded; |
| 1395 | } |
| 1396 | } |
| 1397 | self.decoder.flush_block() |
| 1398 | } |
| 1399 | } |
| 1400 | |
| 1401 | impl<R: BufRead> Iterator for Reader<R> { |
no test coverage detected