Method AdvanceAndMemoize

cpp/src/arrow/acero/asof_join_node.cc:658–702

Advance the data to be immediately past the tolerance's horizon for the specified timestamp, update latest_time and latest_ref_row to the value that immediately pass the horizon. Update the memo-store with any entries or future entries so observed. Returns true if updates were made, false if not. NOTE: The emptiness must be decided by a single call to Empty() in caller, due to the potential race w

  // NOTE: The emptiness must be decided by a single call to Empty() in caller, due to the
  // potential race with Push(), see GH-41614.
  Result<bool> AdvanceAndMemoize(OnType ts, bool empty) {
    // Advance the right side row index until we reach the latest right row (for each key)
    // for the given left timestamp.
    DEBUG_SYNC(node_, "Advancing input ", index_, DEBUG_MANIP(std::endl));

    // Check if already updated for TS (or if there is no latest)
    if (empty) {  // can't advance if empty and no future entries
      return memo_.no_future_ ? false : memo_.RemoveEntriesWithLesserTime(ts);
    }

    // Not updated.  Try to update and possibly advance.
    bool advanced, updated = false;
    OnType latest_time;
    do {
      latest_time = GetLatestTime();
      // if Advance() returns true, then the latest_ts must also be valid
      // Keep advancing right table until we hit the latest row that has
      // timestamp <= ts. This is because we only need the latest row for the
      // match given a left ts.
      if (latest_time > tolerance_.Horizon(ts)) {  // hit a distant timestamp
        DEBUG_SYNC(node_, "Advancing input ", index_, " hit distant time=", latest_time,
                   " at=", ts, DEBUG_MANIP(std::endl));
        // if no future entries, which would have been earlier than the distant time, no
        // need to queue it
        if (memo_.future_entries_.empty()) break;
      }
      auto rb = GetLatestBatch();
      if (may_rehash_ && rb->column_data(key_col_index_[0])->GetNullCount() > 0) {
        must_hash_ = true;
        may_rehash_ = false;
        Rehash();
      }
      memo_.Store(rb, latest_ref_row_, latest_time, DEBUG_ADD(GetLatestKey(), ts));
      // negative tolerance means a last-known entry was stored - set `updated` to `true`
      updated = memo_.no_future_;
      ARROW_ASSIGN_OR_RAISE(advanced, Advance());
    } while (advanced);
    if (!memo_.no_future_ && latest_time >= ts) {
      // `updated` was not modified in the loop from the initial `false` value; set it now
      updated = memo_.RemoveEntriesWithLesserTime(ts);
    }
    DEBUG_SYNC(node_, "Advancing input ", index_, " updated=", updated,
               DEBUG_MANIP(std::endl));
    return updated;
  }
  Status InsertBatch(ExecBatch batch) {
    return sequencer_->InsertBatch(std::move(batch));
  }