Create an instance of `DiscoveryImplIterator` under the hood for the specified directory, wrap it in the `BackgroundGenerator` and feed the results to the main producer queue. Each `DiscoveryImplIterator` maintains a reference to `DiscoveryState`, which simply wraps the producer to keep it alive for the lifetime of this iterator. When all references to `DiscoveryState` are invalidated, the prod
| 519 | /// of this iterator. When all references to `DiscoveryState` are invalidated, |
| 520 | /// the producer is closed automatically. |
| 521 | static Status DoDiscovery(const PlatformFilename& dir_fn, int32_t nesting_depth, |
| 522 | FileSelector selector, |
| 523 | std::shared_ptr<DiscoveryState> discovery_state, |
| 524 | const io::IOContext& io_context, |
| 525 | int32_t file_info_batch_size) { |
| 526 | ARROW_RETURN_IF(discovery_state->producer.is_closed(), |
| 527 | arrow::Status::Cancelled("Discovery cancelled")); |
| 528 | |
| 529 | // Note, that here we use `MakeTransferredGenerator()` with the same |
| 530 | // target executor (io executor) as the current iterator is running on. |
| 531 | // |
| 532 | // This is done on purpose, since typically the user of |
| 533 | // `GetFileInfoGenerator()` would want to perform some more IO on the |
| 534 | // produced results (e.g. read the files, examine metadata etc.). |
| 535 | // So, it is preferable to execute the attached continuations on the same |
| 536 | // executor, which belongs to the IO thread pool. |
| 537 | ARROW_ASSIGN_OR_RAISE( |
| 538 | auto gen, |
| 539 | MakeBackgroundGenerator(Iterator<FileInfoVector>(DiscoveryImplIterator( |
| 540 | dir_fn, nesting_depth, std::move(selector), |
| 541 | discovery_state, io_context, file_info_batch_size)), |
| 542 | io_context.executor())); |
| 543 | gen = MakeTransferredGenerator(std::move(gen), io_context.executor()); |
| 544 | ARROW_RETURN_IF(!discovery_state->producer.Push(std::move(gen)), |
| 545 | arrow::Status::Cancelled("Discovery cancelled")); |
| 546 | return arrow::Status::OK(); |
| 547 | } |
| 548 | }; |
| 549 | |
| 550 | } // anonymous namespace |