Function Emplace

cpp/src/arrow/acero/asof_join_node.cc:837–893

Adds the latest row from the input state as a new composite reference row - LHS must have a valid key,timestep,and latest rows - RHS must have valid data memo'ed for the key

  // - LHS must have a valid key,timestep,and latest rows
  // - RHS must have valid data memo'ed for the key
  void Emplace(std::vector<std::unique_ptr<InputState>>& in, TolType tolerance) {
    DCHECK_EQ(in.size(), n_tables_);

    // Get the LHS key
    ByType key = in[0]->GetLatestKey();

    // Add row and setup LHS
    // (the LHS state comes just from the latest row of the LHS table)
    DCHECK(!in[0]->Empty());
    const std::shared_ptr<arrow::RecordBatch>& lhs_latest_batch = in[0]->GetLatestBatch();
    row_index_t lhs_latest_row = in[0]->GetLatestRow();
    OnType lhs_latest_time = in[0]->GetLatestTime();
    if (0 == lhs_latest_row) {
      // On the first row of the batch, we resize the destination.
      // The destination size is dictated by the size of the LHS batch.
      row_index_t new_batch_size = lhs_latest_batch->num_rows();
      row_index_t new_capacity = unmaterialized_table.Size() + new_batch_size;
      if (unmaterialized_table.capacity() < new_capacity) {
        unmaterialized_table.reserve(new_capacity);
      }
    }

    SliceBuilder new_row{&unmaterialized_table};

    // Each item represents a portion of the columns of the output table
    new_row.AddEntry(lhs_latest_batch, lhs_latest_row, lhs_latest_row + 1);

    DEBUG_SYNC(node_, "Emplace: key=", key, " lhs_latest_row=", lhs_latest_row,
               " lhs_latest_time=", lhs_latest_time, DEBUG_MANIP(std::endl));

    // Get the state for that key from all on the RHS -- assumes it's up to date
    // (the RHS state comes from the memoized row references)
    for (size_t i = 1; i < in.size(); ++i) {
      std::optional<const MemoStore::Entry*> opt_entry = in[i]->GetMemoEntryForKey(key);
#ifndef NDEBUG
      {
        bool has_entry = opt_entry.has_value();
        OnType entry_time = has_entry ? (*opt_entry)->time : TolType::kMinValue;
        row_index_t entry_row = has_entry ? (*opt_entry)->row : 0;
        bool accepted = has_entry && tolerance.Accepts(lhs_latest_time, entry_time);
        DEBUG_SYNC(node_, "  i=", i, " has_entry=", has_entry, " time=", entry_time,
                   " row=", entry_row, " accepted=", accepted, DEBUG_MANIP(std::endl));
      }
#endif
      if (opt_entry.has_value()) {
        DCHECK(*opt_entry);
        if (tolerance.Accepts(lhs_latest_time, (*opt_entry)->time)) {
          // Have a valid entry
          const MemoStore::Entry* entry = *opt_entry;
          new_row.AddEntry(entry->batch, entry->row, entry->row + 1);
          continue;
        }
      }
      new_row.AddEntry(nullptr, 0, 1);
    }
    new_row.Finalize();
  }