| 1194 | } |
| 1195 | |
| 1196 | void FetchNext() { |
| 1197 | size_t row_group_index = readahead_index_++; |
| 1198 | int row_group = row_groups_[row_group_index]; |
| 1199 | std::vector<int> column_indices = column_indices_; |
| 1200 | auto reader = arrow_reader_; |
| 1201 | int64_t num_rows = |
| 1202 | reader->parquet_reader()->metadata()->RowGroup(row_group)->num_rows(); |
| 1203 | rows_in_flight_ += num_rows; |
| 1204 | ::arrow::Future<RecordBatchGenerator> row_group_read; |
| 1205 | if (!reader->properties().pre_buffer()) { |
| 1206 | row_group_read = SubmitRead(cpu_executor_, reader, row_group, column_indices); |
| 1207 | } else { |
| 1208 | auto ready = reader->parquet_reader()->WhenBuffered({row_group}, column_indices); |
| 1209 | if (cpu_executor_) ready = cpu_executor_->TransferAlways(ready); |
| 1210 | row_group_read = |
| 1211 | ready.Then([cpu_executor = cpu_executor_, reader, row_group, |
| 1212 | column_indices = std::move( |
| 1213 | column_indices)]() -> ::arrow::Future<RecordBatchGenerator> { |
| 1214 | return ReadOneRowGroup(cpu_executor, reader, row_group, column_indices); |
| 1215 | }); |
| 1216 | } |
| 1217 | in_flight_reads_.push({std::move(row_group_read), num_rows}); |
| 1218 | } |
| 1219 | |
| 1220 | // Synchronous fallback for when pre-buffer isn't enabled. |
| 1221 | // |
nothing calls this directly
no test coverage detected