| 711 | return Push(rb); |
| 712 | } |
| 713 | void Rehash() { |
| 714 | DEBUG_SYNC(node_, "rehashing for input ", index_, ":", DEBUG_MANIP(std::endl)); |
| 715 | MemoStore new_memo(DEBUG_ADD(memo_.no_future_, node_, index_)); |
| 716 | new_memo.current_time_ = (OnType)memo_.current_time_; |
| 717 | for (auto e = memo_.entries_.begin(); e != memo_.entries_.end(); ++e) { |
| 718 | auto& entry = e->second; |
| 719 | auto new_key = GetKey(entry.batch.get(), entry.row); |
| 720 | DEBUG_SYNC(node_, " ", e->first, " to ", new_key, DEBUG_MANIP(std::endl)); |
| 721 | new_memo.entries_[new_key].swap(entry); |
| 722 | auto fe = memo_.future_entries_.find(e->first); |
| 723 | if (fe != memo_.future_entries_.end()) { |
| 724 | new_memo.future_entries_[new_key].swap(fe->second); |
| 725 | } |
| 726 | } |
| 727 | memo_.times_.swap(new_memo.times_); |
| 728 | memo_.swap(new_memo); |
| 729 | } |
| 730 | |
| 731 | Status Push(const std::shared_ptr<arrow::RecordBatch>& rb) { |
| 732 | if (rb->num_rows() > 0) { |