Loads the batch with the given ID (first row offset) from the inner reader After this call the required batch will be available in `self.local_cache` and may also be stored in `self.shared_cache`.
(&mut self, batch_id: BatchID)
| 127 | /// `self.local_cache` and may also be stored in `self.shared_cache`. |
| 128 | /// |
| 129 | fn fetch_batch(&mut self, batch_id: BatchID) -> Result<usize> { |
| 130 | let first_row_offset = batch_id.val * self.batch_size; |
| 131 | if self.inner_position < first_row_offset { |
| 132 | let to_skip = first_row_offset - self.inner_position; |
| 133 | let skipped = self.inner.skip_records(to_skip)?; |
| 134 | assert_eq!(skipped, to_skip); |
| 135 | self.inner_position += skipped; |
| 136 | } |
| 137 | |
| 138 | let read = self.inner.read_records(self.batch_size)?; |
| 139 | |
| 140 | // If there are no remaining records (EOF), return immediately without |
| 141 | // attempting to cache an empty batch. This prevents inserting zero-length |
| 142 | // arrays into the cache which can later cause panics when slicing. |
| 143 | if read == 0 { |
| 144 | return Ok(0); |
| 145 | } |
| 146 | |
| 147 | let array = self.inner.consume_batch()?; |
| 148 | |
| 149 | // Store in both shared cache and local cache |
| 150 | // The shared cache is used to reuse results between readers |
| 151 | // The local cache ensures data is available for our consume_batch call |
| 152 | let _cached = |
| 153 | self.shared_cache |
| 154 | .write() |
| 155 | .unwrap() |
| 156 | .insert(self.column_idx, batch_id, array.clone()); |
| 157 | // Note: if the shared cache is full (_cached == false), we continue without caching |
| 158 | // The local cache will still store the data for this reader's use |
| 159 | |
| 160 | self.local_cache.insert(batch_id, array); |
| 161 | |
| 162 | self.inner_position += read; |
| 163 | Ok(read) |
| 164 | } |
| 165 | |
| 166 | /// Remove batches from cache that have been completely consumed |
| 167 | /// This is only called for Consumer role readers |
no test coverage detected