Method RemoveEntriesWithLesserTime

cpp/src/arrow/acero/asof_join_node.cc:343–384

  }

  bool RemoveEntriesWithLesserTime(OnType ts) {
    DEBUG_SYNC(node_, "memo ", index_, " remove: ts=", ts, DEBUG_MANIP(std::endl));
    bool updated = false;
    // remove future entries with lesser time
    for (auto fe = future_entries_.begin(); fe != future_entries_.end();) {
      auto& queue = fe->second;
      while (!queue.empty() && queue.front().time < ts) {
        queue.pop();
        updated = true;  // queue changed
      }
      // remove entry if its queue was just emptied
      if (queue.empty()) {
        fe = future_entries_.erase(fe);
      } else {
        ++fe;
      }
    }
    // remove last-known entries with lesser time
    for (auto e = entries_.begin(); e != entries_.end();) {
      if (e->second.time < ts) {
        // drop last-known entry and move next future entry, if exists, in its place
        auto fe = future_entries_.find(e->first);
        if (fe != future_entries_.end() && !fe->second.empty()) {
          auto& queue = fe->second;
          e->second.swap(queue.front());
          queue.pop();
          ++e;
        } else {
          e = entries_.erase(e);
        }
        updated = true;  // entry changed
      } else {
        ++e;
      }
    }
    // remove known lesser times
    while (!times_.empty() && times_.front() < ts) {
      times_.pop_front();
    }
    // update current time
    return UpdateTime(ts) || updated;
  }
};

// a specialized higher-performance variation of Hashing64 logic from hash_join_node