lbug 0.16.1

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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

#pragma once

#include "common/in_mem_overflow_buffer.h"
#include "common/types/int128_t.h"
#include "common/types/uint128_t.h"
#include "function/aggregate_function.h"
#include "function/arithmetic/add.h"

namespace lbug {
namespace function {

template<typename T>
struct AvgState : public AggregateStateWithNull {
    uint32_t getStateSize() const override { return sizeof(*this); }
    void writeToVector(common::ValueVector* outputVector, uint64_t pos) override {
        outputVector->setValue(pos, avg);
    }

    void finalize()
        requires common::IntegerTypes<T>
    {
        using ResultType = std::conditional<common::SignedIntegerTypes<T>, common::Int128_t,
            common::UInt128_t>::type;
        if (!isNull) {
            avg = ResultType::template cast<long double>(sum) /
                  ResultType::template cast<long double>(count);
        }
    }

    void finalize()
        requires common::FloatingPointTypes<T>
    {
        if (!isNull) {
            avg = sum / count;
        }
    }

    T sum{};
    uint64_t count = 0;
    double avg = 0;
};

template<typename INPUT_TYPE, typename RESULT_TYPE>
struct AvgFunction {

    static std::unique_ptr<AggregateState> initialize() {
        return std::make_unique<AvgState<RESULT_TYPE>>();
    }

    static void updateAll(uint8_t* state_, common::ValueVector* input, uint64_t multiplicity,
        common::InMemOverflowBuffer* /*overflowBuffer*/) {
        auto* state = reinterpret_cast<AvgState<RESULT_TYPE>*>(state_);
        DASSERT(!input->state->isFlat());
        input->forEachNonNull(
            [&](auto pos) { updateSingleValue(state, input, pos, multiplicity); });
    }

    static void updatePos(uint8_t* state_, common::ValueVector* input, uint64_t multiplicity,
        uint32_t pos, common::InMemOverflowBuffer* /*overflowBuffer*/) {
        updateSingleValue(reinterpret_cast<AvgState<RESULT_TYPE>*>(state_), input, pos,
            multiplicity);
    }

    static void updateSingleValue(AvgState<RESULT_TYPE>* state, common::ValueVector* input,
        uint32_t pos, uint64_t multiplicity) {
        INPUT_TYPE val = input->getValue<INPUT_TYPE>(pos);
        for (auto i = 0u; i < multiplicity; ++i) {
            if (state->isNull) {
                state->sum = (RESULT_TYPE)val;
                state->isNull = false;
            } else {
                Add::operation(state->sum, val, state->sum);
            }
        }
        state->count += multiplicity;
    }

    static void combine(uint8_t* state_, uint8_t* otherState_,
        common::InMemOverflowBuffer* /*overflowBuffer*/) {
        auto* otherState = reinterpret_cast<AvgState<RESULT_TYPE>*>(otherState_);
        if (otherState->isNull) {
            return;
        }
        auto* state = reinterpret_cast<AvgState<RESULT_TYPE>*>(state_);
        if (state->isNull) {
            state->sum = otherState->sum;
            state->isNull = false;
        } else {
            Add::operation(state->sum, otherState->sum, state->sum);
        }
        state->count = state->count + otherState->count;
    }

    static void finalize(uint8_t* state_) {
        auto* state = reinterpret_cast<AvgState<RESULT_TYPE>*>(state_);
        state->finalize();
    }
};

} // namespace function
} // namespace lbug