Method GetNextMaterialized

be/src/exec/union-node.cc:210–282

}

Status UnionNode::GetNextMaterialized(RuntimeState* state, RowBatch* row_batch) {
  // Fetch from children, evaluate corresponding exprs and materialize.
  DCHECK(!ReachedLimit());
  DCHECK_LT(child_idx_, children_.size());
  int64_t tuple_buf_size;
  uint8_t* tuple_buf;
  RETURN_IF_ERROR(
      row_batch->ResizeAndAllocateTupleBuffer(state, &tuple_buf_size, &tuple_buf));
  memset(tuple_buf, 0, tuple_buf_size);

  while (HasMoreMaterialized() && !row_batch->AtCapacity()) {
    // The loop runs until we are either done iterating over the children that require
    // materialization, or the row batch is at capacity.
    DCHECK(!IsChildPassthrough(child_idx_));
    // Child row batch was either never set or we're moving on to a different child.
    if (child_batch_.get() == nullptr) {
      DCHECK_LT(child_idx_, children_.size());
      child_batch_.reset(new RowBatch(
          child(child_idx_)->row_desc(), state->batch_size(), mem_tracker()));
      child_row_idx_ = 0;
      // Open the current child unless it's the first child, which was already opened in
      // UnionNode::Open().
      if (child_eos_) RETURN_IF_ERROR(child(child_idx_)->Open(state));
      // The first batch from each child is always fetched here.
      RETURN_IF_ERROR(child(child_idx_)->GetNext(
          state, child_batch_.get(), &child_eos_));
    }

    while (!row_batch->AtCapacity()) {
      DCHECK(child_batch_.get() != nullptr);
      DCHECK_LE(child_row_idx_, child_batch_->num_rows());
      if (child_row_idx_ == child_batch_->num_rows()) {
        // Move on to the next child if it is at eos.
        if (child_eos_) break;
        // Fetch more rows from the child.
        child_batch_->Reset();
        child_row_idx_ = 0;
        // All batches except the first batch from each child are fetched here.
        RETURN_IF_ERROR(child(child_idx_)->GetNext(
            state, child_batch_.get(), &child_eos_));
        // If we fetched an empty batch, go back to the beginning of this while loop, and
        // try again.
        if (child_batch_->num_rows() == 0) continue;
      }
      DCHECK_EQ(codegend_union_materialize_batch_fns_.size(), children_.size());
      UnionPlanNode::UnionMaterializeBatchFn fn =
          codegend_union_materialize_batch_fns_[child_idx_].load();
      if (fn == nullptr) {
        MaterializeBatch(row_batch, &tuple_buf);
      } else {
        fn(this, row_batch, &tuple_buf);
      }
    }
    // It shouldn't be the case that we reached the limit because we shouldn't have
    // incremented 'num_rows_returned_' yet.
    DCHECK(!ReachedLimit());

    if (child_eos_ && child_row_idx_ == child_batch_->num_rows()) {