| 468 | |
| 469 | private: |
| 470 | static Future<DecodedBlock> FirstBlock(AsyncGenerator<DecodedBlock> gen) { |
| 471 | // Read from the stream until we get a non-empty record batch that we can use to |
| 472 | // declare the schema. Along the way, accumulate the bytes read so they can be |
| 473 | // recorded on the first `ReadNextAsync` |
| 474 | auto loop_body = |
| 475 | [gen = std::move(gen), |
| 476 | out = std::make_shared<DecodedBlock>()]() -> Future<ControlFlow<DecodedBlock>> { |
| 477 | return gen().Then( |
| 478 | [out](const DecodedBlock& block) -> Result<ControlFlow<DecodedBlock>> { |
| 479 | if (IsIterationEnd(block)) { |
| 480 | return Status::Invalid("Empty JSON stream"); |
| 481 | } |
| 482 | out->num_bytes += block.num_bytes; |
| 483 | if (block.record_batch->num_rows() == 0) { |
| 484 | return Continue(); |
| 485 | } |
| 486 | out->record_batch = block.record_batch; |
| 487 | return Break(*out); |
| 488 | }); |
| 489 | }; |
| 490 | return Loop(std::move(loop_body)); |
| 491 | } |
| 492 | |
| 493 | std::optional<DecodedBlock> first_block_; |
| 494 | std::shared_ptr<Schema> schema_; |