Method ProcessInner

cpp/src/arrow/acero/asof_join_node.cc:979–1027

  }

  Result<std::shared_ptr<RecordBatch>> ProcessInner() {
    DCHECK(!state_.empty());
    auto& lhs = *state_.at(0);

    // Construct new target table if needed
    CompositeTableBuilder<MAX_JOIN_TABLES> dst(state_, output_schema_,
                                               plan()->query_context()->memory_pool(),
                                               DEBUG_ADD(state_.size(), this));

    // Generate rows into the dst table until we either run out of data or hit the row
    // limit, or run out of input
    for (;;) {
      // If LHS is finished or empty then there's nothing we can do here
      if (lhs.Finished() || lhs.Empty()) break;

      ARROW_ASSIGN_OR_RAISE(auto rhs_update_state, UpdateRhs());

      // If we have received enough inputs to produce the next output batch
      // (decided by IsUpToDateWithLhsRow), we will perform the join and
      // materialize the output batch. The join is done by advancing through
      // the LHS and adding joined row to rows_ (done by Emplace). Finally,
      // input batches that are no longer needed are removed to free up memory.
      if (rhs_update_state.all_up_to_date_with_lhs) {
        dst.Emplace(state_, tolerance_);
        ARROW_ASSIGN_OR_RAISE(bool advanced, lhs.Advance());
        if (!advanced) break;  // if we can't advance LHS, we're done for this batch
      } else {
        if (!rhs_update_state.any_advanced) break;  // need to wait for new data
      }
    }

    // Prune memo entries that have expired (to bound memory consumption)
    if (!lhs.Empty()) {
      for (size_t i = 1; i < state_.size(); ++i) {
        OnType ts = tolerance_.Expiry(lhs.GetLatestTime());
        if (ts != TolType::kMinValue) {
          state_[i]->RemoveMemoEntriesWithLesserTime(ts);
        }
      }
    }

    // Emit the batch
    if (dst.empty()) {
      return NULLPTR;
    } else {
      ARROW_ASSIGN_OR_RAISE(auto out, dst.Materialize());
      return out.has_value() ? out.value() : NULLPTR;
    }
  }

#ifdef ARROW_ENABLE_THREADING