Method GetNext

be/src/exec/unpivot-node.cc:138–187

Like a UnionNode, an UnpivotNode materializes tuples by evaluating expressions. Unlike a UnionNode, for each input tuple, an UnpivotNode might produce multiple output tuples.

// Unlike a UnionNode, for each input tuple, an UnpivotNode might produce multiple
// output tuples.
Status UnpivotNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  RETURN_IF_CANCELLED(state);
  RETURN_IF_ERROR(QueryMaintenance(state));
  const auto& plan_node = static_cast<const UnpivotPlanNode&>(plan_node_);
  int64_t tuple_buf_size = -1;
  uint8_t* tuple_buf = nullptr;
  RETURN_IF_ERROR(
      row_batch->ResizeAndAllocateTupleBuffer(state, &tuple_buf_size, &tuple_buf));
  memset(tuple_buf, 0, tuple_buf_size);
  auto output_tuple = tuple_buf;
  DCHECK_GT(plan_node.num_unpivot_columns_, 0);
  // Materializes one row in each iteration if *eos is false.
  do {
    if (unpivot_slot_idx_ == plan_node.num_unpivot_columns_) {
      ++child_row_idx_;
      unpivot_slot_idx_ = 0;
    }
    if (child_row_batch_ == nullptr || child_row_idx_ == child_row_batch_->num_rows()) {
      if (child_row_batch_ == nullptr) {
        child_row_batch_ = make_unique<RowBatch>(
            child(0)->row_desc(), state->batch_size(), mem_tracker());
      } else {
        child_row_batch_->Reset();
      }
      if (!child_eos_) {
        RETURN_IF_ERROR(child(0)->GetNext(state, child_row_batch_.get(), &child_eos_));
      }
      child_row_idx_ = 0;
    }
    *eos = ReachedLimit()
        || (child_eos_ && (child_row_idx_ == child_row_batch_->num_rows()));
    if (*eos) {
      return Status::OK();
    }
    if (child_row_batch_->num_rows() == 0) {
      continue;
    }
    MaterializeOutputTuple(reinterpret_cast<Tuple*>(output_tuple), row_batch);
    ++unpivot_slot_idx_;
    auto output_row = row_batch->GetRow(row_batch->AddRow());
    output_row->SetTuple(0, reinterpret_cast<Tuple*>(output_tuple));
    if (EvalConjuncts(conjunct_evals_.data(), conjunct_evals_.size(), output_row)) {
      row_batch->CommitLastRow();
      IncrementNumRowsReturned(1);
      output_tuple += plan_node.tuple_desc_->byte_size();
    }
  } while (!*eos && !row_batch->AtCapacity());
  return Status::OK();
}

Status UnpivotNode::Reset(RuntimeState* state, RowBatch* row_batch) {
  if (child_row_batch_ != nullptr) {