| 278 | } |
| 279 | |
| 280 | static inline Future<std::shared_ptr<csv::StreamingReader>> OpenReaderAsync( |
| 281 | const FileSource& source, const CsvFileFormat& format, |
| 282 | const std::shared_ptr<ScanOptions>& scan_options, Executor* cpu_executor) { |
| 283 | #ifdef ARROW_WITH_OPENTELEMETRY |
| 284 | auto tracer = arrow::internal::tracing::GetTracer(); |
| 285 | auto span = tracer->StartSpan("arrow::dataset::CsvFileFormat::OpenReaderAsync"); |
| 286 | #endif |
| 287 | ARROW_ASSIGN_OR_RAISE( |
| 288 | auto fragment_scan_options, |
| 289 | GetFragmentScanOptions<CsvFragmentScanOptions>( |
| 290 | kCsvTypeName, scan_options.get(), format.default_fragment_scan_options)); |
| 291 | ARROW_ASSIGN_OR_RAISE(auto reader_options, GetReadOptions(format, scan_options)); |
| 292 | ARROW_ASSIGN_OR_RAISE(auto input, source.OpenCompressed()); |
| 293 | if (fragment_scan_options->stream_transform_func) { |
| 294 | ARROW_ASSIGN_OR_RAISE(input, fragment_scan_options->stream_transform_func(input)); |
| 295 | } |
| 296 | const auto& path = source.path(); |
| 297 | ARROW_ASSIGN_OR_RAISE( |
| 298 | input, io::BufferedInputStream::Create(reader_options.block_size, |
| 299 | default_memory_pool(), std::move(input))); |
| 300 | |
| 301 | // Grab the first block and use it to determine the schema and create a reader. The |
| 302 | // input->Peek call blocks so we run the whole thing on the I/O thread pool. |
| 303 | auto reader_fut = DeferNotOk(input->io_context().executor()->Submit( |
| 304 | [=]() -> Future<std::shared_ptr<csv::StreamingReader>> { |
| 305 | ARROW_ASSIGN_OR_RAISE(auto first_block, input->Peek(reader_options.block_size)); |
| 306 | const auto& parse_options = format.parse_options; |
| 307 | ARROW_ASSIGN_OR_RAISE( |
| 308 | auto convert_options, |
| 309 | GetConvertOptions(format, scan_options ? scan_options.get() : nullptr, |
| 310 | first_block)); |
| 311 | return csv::StreamingReader::MakeAsync(io::default_io_context(), std::move(input), |
| 312 | cpu_executor, reader_options, |
| 313 | parse_options, convert_options); |
| 314 | })); |
| 315 | return reader_fut.Then( |
| 316 | // Adds the filename to the error |
| 317 | [=](const std::shared_ptr<csv::StreamingReader>& reader) |
| 318 | -> Result<std::shared_ptr<csv::StreamingReader>> { |
| 319 | #ifdef ARROW_WITH_OPENTELEMETRY |
| 320 | span->SetStatus(opentelemetry::trace::StatusCode::kOk); |
| 321 | span->End(); |
| 322 | #endif |
| 323 | return reader; |
| 324 | }, |
| 325 | [=](const Status& err) -> Result<std::shared_ptr<csv::StreamingReader>> { |
| 326 | #ifdef ARROW_WITH_OPENTELEMETRY |
| 327 | arrow::internal::tracing::MarkSpan(err, span.get()); |
| 328 | span->End(); |
| 329 | #endif |
| 330 | return err.WithMessage("Could not open CSV input source '", path, "': ", err); |
| 331 | }); |
| 332 | } |
| 333 | |
| 334 | static inline Result<std::shared_ptr<csv::StreamingReader>> OpenReader( |
| 335 | const FileSource& source, const CsvFileFormat& format, |
no test coverage detected