| 155 | } |
| 156 | |
| 157 | Result<RecordBatchGenerator> OrcFileFormat::ScanBatchesAsync( |
| 158 | const std::shared_ptr<ScanOptions>& options, |
| 159 | const std::shared_ptr<FileFragment>& file) const { |
| 160 | // TODO investigate "true" async version |
| 161 | // (https://issues.apache.org/jira/browse/ARROW-13795) |
| 162 | ARROW_ASSIGN_OR_RAISE(auto task_iter, OrcScanTaskIterator::Make(options, file)); |
| 163 | struct IterState { |
| 164 | Iterator<std::shared_ptr<OrcScanTask>> iter; |
| 165 | RecordBatchIterator curr_iter; |
| 166 | bool first; |
| 167 | ::arrow::internal::Executor* io_executor; |
| 168 | }; |
| 169 | struct { |
| 170 | Future<std::shared_ptr<RecordBatch>> operator()() { |
| 171 | auto state = state_; |
| 172 | return ::arrow::DeferNotOk( |
| 173 | state->io_executor->Submit([state]() -> Result<std::shared_ptr<RecordBatch>> { |
| 174 | if (state->first) { |
| 175 | ARROW_ASSIGN_OR_RAISE(auto task, state->iter.Next()); |
| 176 | ARROW_ASSIGN_OR_RAISE(state->curr_iter, task->Execute()); |
| 177 | state->first = false; |
| 178 | } |
| 179 | while (!IsIterationEnd(state->curr_iter)) { |
| 180 | ARROW_ASSIGN_OR_RAISE(auto next_batch, state->curr_iter.Next()); |
| 181 | if (IsIterationEnd(next_batch)) { |
| 182 | ARROW_ASSIGN_OR_RAISE(auto task, state->iter.Next()); |
| 183 | if (IsIterationEnd(task)) { |
| 184 | state->curr_iter = IterationEnd<RecordBatchIterator>(); |
| 185 | } else { |
| 186 | ARROW_ASSIGN_OR_RAISE(state->curr_iter, task->Execute()); |
| 187 | } |
| 188 | } else { |
| 189 | return next_batch; |
| 190 | } |
| 191 | } |
| 192 | return IterationEnd<std::shared_ptr<RecordBatch>>(); |
| 193 | })); |
| 194 | } |
| 195 | std::shared_ptr<IterState> state_; |
| 196 | } iter_to_gen{std::shared_ptr<IterState>( |
| 197 | new IterState{std::move(task_iter), {}, true, options->io_context.executor()})}; |
| 198 | return iter_to_gen; |
| 199 | } |
| 200 | |
| 201 | Future<std::optional<int64_t>> OrcFileFormat::CountRows( |
| 202 | const std::shared_ptr<FileFragment>& file, compute::Expression predicate, |