| 258 | } |
| 259 | |
| 260 | Result<Future<ReaderPtr>> DoOpenReader( |
| 261 | const FileSource& source, const JsonFileFormat& format, |
| 262 | const std::shared_ptr<ScanOptions>& scan_options = nullptr) { |
| 263 | ARROW_ASSIGN_OR_RAISE(auto json_options, |
| 264 | GetJsonFormatOptions(format, scan_options.get())); |
| 265 | |
| 266 | struct State { |
| 267 | State(const JsonFragmentScanOptions& json_options, |
| 268 | const std::shared_ptr<ScanOptions>& scan_options) |
| 269 | : parse_options(GetInitialParseOptions(json_options.parse_options)), |
| 270 | read_options(json_options.read_options), |
| 271 | scan_options(scan_options) {} |
| 272 | json::ParseOptions parse_options; |
| 273 | json::ReadOptions read_options; |
| 274 | std::shared_ptr<const ScanOptions> scan_options; |
| 275 | std::shared_ptr<io::InputStream> stream; |
| 276 | }; |
| 277 | |
| 278 | auto state = std::make_shared<State>(*json_options, scan_options); |
| 279 | ARROW_ASSIGN_OR_RAISE(state->stream, source.OpenCompressed()); |
| 280 | ARROW_ASSIGN_OR_RAISE( |
| 281 | state->stream, |
| 282 | io::BufferedInputStream::Create(state->read_options.block_size, |
| 283 | default_memory_pool(), std::move(state->stream))); |
| 284 | |
| 285 | auto maybe_future = state->stream->io_context().executor()->Submit( |
| 286 | [state = std::move(state)]() -> Future<ReaderPtr> { |
| 287 | if (state->scan_options && state->scan_options->dataset_schema) { |
| 288 | // Inspect the first block before anything else, so we can derive an explicit |
| 289 | // schema for the reader based on the dataset schema. |
| 290 | ARROW_ASSIGN_OR_RAISE(auto first_block, |
| 291 | state->stream->Peek(state->read_options.block_size)); |
| 292 | ARROW_ASSIGN_OR_RAISE(auto physical_schema, |
| 293 | ParseToSchema(first_block, state->parse_options, |
| 294 | state->scan_options->pool)); |
| 295 | ARROW_ASSIGN_OR_RAISE(state->parse_options.explicit_schema, |
| 296 | GetPartialSchema(*state->scan_options, *physical_schema)); |
| 297 | state->parse_options.unexpected_field_behavior = |
| 298 | json::UnexpectedFieldBehavior::Ignore; |
| 299 | } |
| 300 | return json::StreamingReader::MakeAsync( |
| 301 | std::move(state->stream), state->read_options, state->parse_options); |
| 302 | }); |
| 303 | ARROW_ASSIGN_OR_RAISE(auto future, maybe_future); |
| 304 | return future.Then([](const ReaderPtr& reader) -> Result<ReaderPtr> { return reader; }, |
| 305 | [path = source.path()](const Status& error) -> Result<ReaderPtr> { |
| 306 | return error.WithMessage("Could not open JSON input source '", |
| 307 | path, "': ", error); |
| 308 | }); |
| 309 | } |
| 310 | |
| 311 | Future<ReaderPtr> OpenReaderAsync( |
| 312 | const FileSource& source, const JsonFileFormat& format, |
no test coverage detected