Method executeInternal

src/processor/operator/aggregate/simple_aggregate.cpp:218–242

}

void SimpleAggregate::executeInternal(ExecutionContext* context) {
    InMemOverflowBuffer localOverflowBuffer(storage::MemoryManager::Get(*context->clientContext));
    while (children[0]->getNextTuple(context)) {
        for (auto i = 0u; i < aggregateFunctions.size(); i++) {
            auto aggregateFunction = &aggregateFunctions[i];
            if (aggregateFunction->isFunctionDistinct()) {
                // Just add distinct value to the hash table. We'll calculate the aggregate state
                // once it's been merged into the shared state
                distinctHashTables[i]->appendDistinct(std::vector<ValueVector*>{},
                    aggInputs[i].aggregateVector, aggInputs[i].aggregateVector->state.get());
            } else {
                computeAggregate(aggregateFunction, &aggInputs[i], localAggregateStates[i].get(),
                    localOverflowBuffer);
            }
        }
    }
    if (getSharedState().hasDistinct) {
        for (auto& hashTable : distinctHashTables) {
            if (hashTable) {
                hashTable->mergeIfFull(0 /*tuplesToAdd*/, true /*mergeAll*/);
            }
        }
    }
    getSharedState().combineAggregateStates(localAggregateStates, std::move(localOverflowBuffer));
}

void SimpleAggregate::computeAggregate(function::AggregateFunction* function, AggregateInput* input,
    function::AggregateState* state, common::InMemOverflowBuffer& overflowBuffer) {