Method mergeInto

src/processor/operator/aggregate/base_aggregate.cpp:81–97

}

void BaseAggregateSharedState::HashTableQueue::mergeInto(AggregateHashTable& hashTable) {
    TupleBlock* partitionToMerge = nullptr;
    auto headBlock = this->headBlock.load();
    DASSERT(headBlock != nullptr);
    while (queuedTuples.pop(partitionToMerge)) {
        DASSERT(
            partitionToMerge->numTuplesWritten == partitionToMerge->table.getNumTuplesPerBlock());
        hashTable.merge(std::move(partitionToMerge->table));
        delete partitionToMerge;
    }
    if (headBlock->numTuplesWritten > 0) {
        headBlock->table.resize(headBlock->numTuplesWritten);
        hashTable.merge(std::move(headBlock->table));
    }
    delete headBlock;
    this->headBlock = nullptr;
}

} // namespace processor
} // namespace lbug