lbug 0.15.4

An in-process property graph database management system built for query speed and scalability
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

#include "processor/operator/aggregate/simple_aggregate_scan.h"

namespace lbug {
namespace processor {

void SimpleAggregateScan::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
    BaseAggregateScan::initLocalStateInternal(resultSet, context);
    DASSERT(!scanInfo.aggregatesPos.empty());
    auto outDataChunkPos = scanInfo.aggregatesPos[0].dataChunkPos;
    RUNTIME_CHECK({
        for (auto& dataPos : scanInfo.aggregatesPos) {
            DASSERT(dataPos.dataChunkPos == outDataChunkPos);
        }
    });
    outDataChunk = resultSet->dataChunks[outDataChunkPos].get();
}

bool SimpleAggregateScan::getNextTuplesInternal(ExecutionContext* /*context*/) {
    auto [startOffset, endOffset] = sharedState->getNextRangeToRead();
    if (startOffset >= endOffset) {
        return false;
    }
    // Output of simple aggregate is guaranteed to be a single value for each aggregate.
    DASSERT(startOffset == 0 && endOffset == 1);
    for (auto i = 0u; i < aggregateVectors.size(); i++) {
        scanInfo.moveAggResultToVectorFuncs[i](*aggregateVectors[i], 0 /* position to write */,
            sharedState->getAggregateState(i));
    }
    DASSERT(!scanInfo.aggregatesPos.empty());
    outDataChunk->state->initOriginalAndSelectedSize(1);
    metrics->numOutputTuple.increase(outDataChunk->state->getSelVector().getSelSize());
    return true;
}

} // namespace processor
} // namespace lbug