| 103 | } |
| 104 | |
| 105 | Status SubplanNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) { |
| 106 | SCOPED_TIMER(runtime_profile_->total_time_counter()); |
| 107 | ScopedGetNextEventAdder ea(this, eos); |
| 108 | RETURN_IF_CANCELLED(state); |
| 109 | RETURN_IF_ERROR(QueryMaintenance(state)); |
| 110 | *eos = false; |
| 111 | |
| 112 | while (true) { |
| 113 | if (subplan_is_open_) { |
| 114 | if (subplan_eos_) { |
| 115 | // Reset the subplan before opening it again. 'row_batch' is passed in to allow |
| 116 | // any remaining resources to be transferred to it. |
| 117 | RETURN_IF_ERROR(child(1)->Reset(state, row_batch)); |
| 118 | subplan_is_open_ = false; |
| 119 | } else { |
| 120 | // Continue fetching rows from the open subplan into the output row_batch. |
| 121 | DCHECK(!row_batch->AtCapacity()); |
| 122 | RETURN_IF_ERROR(child(1)->GetNext(state, row_batch, &subplan_eos_)); |
| 123 | // Apply limit and check whether the output batch is at capacity. |
| 124 | if (limit_ != -1 && rows_returned() + row_batch->num_rows() >= limit_) { |
| 125 | row_batch->set_num_rows(limit_ - rows_returned()); |
| 126 | IncrementNumRowsReturned(row_batch->num_rows()); |
| 127 | *eos = true; |
| 128 | break; |
| 129 | } |
| 130 | if (row_batch->AtCapacity()) { |
| 131 | IncrementNumRowsReturned(row_batch->num_rows()); |
| 132 | return Status::OK(); |
| 133 | } |
| 134 | // Check subplan_eos_ and repeat fetching until the output batch is at capacity |
| 135 | // or we have reached our limit. |
| 136 | continue; |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | if (input_row_idx_ >= input_batch_->num_rows()) { |
| 141 | input_batch_->TransferResourceOwnership(row_batch); |
| 142 | if (input_eos_) { |
| 143 | *eos = true; |
| 144 | break; |
| 145 | } |
| 146 | // Could be at capacity after resources have been transferred to it. |
| 147 | if (row_batch->AtCapacity()) return Status::OK(); |
| 148 | // Continue fetching input rows. |
| 149 | input_batch_->Reset(); |
| 150 | RETURN_IF_ERROR(child(0)->GetNext(state, input_batch_.get(), &input_eos_)); |
| 151 | input_row_idx_ = 0; |
| 152 | if (input_batch_->num_rows() == 0) continue; |
| 153 | } |
| 154 | |
| 155 | // Advance the current input row to be picked up by dependent nodes, |
| 156 | // and Open() the subplan. |
| 157 | current_input_row_ = input_batch_->GetRow(input_row_idx_); |
| 158 | ++input_row_idx_; |
| 159 | RETURN_IF_ERROR(child(1)->Open(state)); |
| 160 | subplan_is_open_ = true; |
| 161 | subplan_eos_ = false; |
| 162 | } |
nothing calls this directly
no test coverage detected