| 587 | } |
| 588 | |
| 589 | bytedance::bolt::RowVectorPtr HashAggregation::getOutput() { |
| 590 | if (finished_ || !noMoreInput_) { |
| 591 | return nullptr; |
| 592 | } |
| 593 | |
| 594 | if (inputs_.size() > 0) { |
| 595 | std::vector<::cudf::table_view> inputViews; |
| 596 | for (auto& input : inputs_) { |
| 597 | inputViews.emplace_back(input->view()); |
| 598 | } |
| 599 | try { |
| 600 | inputTable_ = ::cudf::concatenate(inputViews, stream_); |
| 601 | } catch (const std::exception& e) { |
| 602 | BOLT_FAIL("HashAggregation failed: {}", e.what()); |
| 603 | } |
| 604 | } else { |
| 605 | BOLT_CHECK_EQ(inputs_.size(), 1, "HashAggregation expects 1 input"); |
| 606 | inputTable_ = std::move(inputs_[0]); |
| 607 | } |
| 608 | |
| 609 | finished_ = true; |
| 610 | |
| 611 | if (isGlobal_) { |
| 612 | runGlobalAggregation(); |
| 613 | } else { |
| 614 | runGroupbyAggregation(); |
| 615 | } |
| 616 | stream_.synchronize(); |
| 617 | |
| 618 | bytedance::bolt::RowVectorPtr outputVector = nullptr; |
| 619 | if (outputTable_->num_rows() != 0 && outputTable_->num_columns() != 0) { |
| 620 | outputVector = toBoltRowVector(outputTable_->view(), pool(), stream_); |
| 621 | outputVector->setType(aggregationNode_->outputType()); |
| 622 | } |
| 623 | |
| 624 | recordInterOpStats(*this, interOpStats_); |
| 625 | |
| 626 | return outputVector; |
| 627 | } |
| 628 | |
| 629 | } // namespace bolt::cudf::exec |
nothing calls this directly
no test coverage detected