Method UpdateRhs

cpp/src/arrow/acero/asof_join_node.cc:948–977

Advances the RHS as far as possible to be up to date for the current LHS timestamp, and checks if all RHS are up to date with LHS. The reason they have to be performed together is that they both depend on the emptiness of the RHS, which can be changed by Push() executing in another thread.

  // together is that they both depend on the emptiness of the RHS, which can be changed
  // by Push() executing in another thread.
  Result<RhsUpdateState> UpdateRhs() {
    auto& lhs = *state_.at(0);
    auto lhs_latest_time = lhs.GetLatestTime();
    RhsUpdateState update_state{/*any_advanced=*/false, /*all_up_to_date_with_lhs=*/true};
    for (size_t i = 1; i < state_.size(); ++i) {
      auto& rhs = *state_[i];

      // Obtain RHS emptiness once for subsequent AdvanceAndMemoize() and CurrentEmpty().
      bool rhs_empty = rhs.Empty();
      // Obtain RHS current time here because AdvanceAndMemoize() can change the
      // emptiness.
      OnType rhs_current_time = rhs_empty ? OnType{} : rhs.GetLatestTime();

      ARROW_ASSIGN_OR_RAISE(bool advanced,
                            rhs.AdvanceAndMemoize(lhs_latest_time, rhs_empty));
      update_state.any_advanced |= advanced;

      if (update_state.all_up_to_date_with_lhs && !rhs.Finished()) {
        // If RHS is finished, then we know it's up to date
        if (rhs.CurrentEmpty(rhs_empty)) {
          // RHS isn't finished, but is empty --> not up to date
          update_state.all_up_to_date_with_lhs = false;
        } else if (lhs_latest_time > rhs_current_time) {
          // RHS isn't up to date (and not finished)
          update_state.all_up_to_date_with_lhs = false;
        }
      }
    }
    return update_state;
  }

  Result<std::shared_ptr<RecordBatch>> ProcessInner() {
    DCHECK(!state_.empty());