| 624 | bool Finished() const { return batches_processed_ == total_batches_; } |
| 625 | |
| 626 | Result<bool> Advance() { |
| 627 | // Try advancing to the next row and update latest_ref_row_ |
| 628 | // Returns true if able to advance, false if not. |
| 629 | bool have_active_batch = |
| 630 | (latest_ref_row_ > 0 /*short circuit the lock on the queue*/) || !queue_.Empty(); |
| 631 | |
| 632 | if (have_active_batch) { |
| 633 | OnType next_time = GetLatestTime(); |
| 634 | if (latest_time_ > next_time) { |
| 635 | return Status::Invalid("AsofJoin does not allow out-of-order on-key values"); |
| 636 | } |
| 637 | latest_time_ = next_time; |
| 638 | // If we have an active batch |
| 639 | if (++latest_ref_row_ >= (row_index_t)queue_.Front()->num_rows()) { |
| 640 | // hit the end of the batch, need to get the next batch if possible. |
| 641 | ++batches_processed_; |
| 642 | latest_ref_row_ = 0; |
| 643 | bool did_pop = queue_.TryPop().has_value(); |
| 644 | DCHECK(did_pop); |
| 645 | ARROW_UNUSED(did_pop); |
| 646 | have_active_batch = !queue_.Empty(); |
| 647 | } |
| 648 | } |
| 649 | return have_active_batch; |
| 650 | } |
| 651 | |
| 652 | // Advance the data to be immediately past the tolerance's horizon for the specified |
| 653 | // timestamp, update latest_time and latest_ref_row to the value that immediately pass |