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

#include "planner/operator/factorization/sink_util.h"

namespace lbug {
namespace planner {

void SinkOperatorUtil::mergeSchema(const Schema& inputSchema,
    const binder::expression_vector& expressionsToMerge, Schema& resultSchema) {
    auto flatPayloads = getFlatPayloads(inputSchema, expressionsToMerge);
    auto unFlatPayloadsPerGroup = getUnFlatPayloadsPerGroup(inputSchema, expressionsToMerge);
    if (unFlatPayloadsPerGroup.empty()) {
        appendPayloadsToNewGroup(resultSchema, flatPayloads);
    } else {
        if (!flatPayloads.empty()) {
            auto groupPos = appendPayloadsToNewGroup(resultSchema, flatPayloads);
            resultSchema.setGroupAsSingleState(groupPos);
        }
        for (auto& [inputGroupPos, payloads] : unFlatPayloadsPerGroup) {
            auto resultGroupPos = appendPayloadsToNewGroup(resultSchema, payloads);
            resultSchema.getGroup(resultGroupPos)
                ->setMultiplier(inputSchema.getGroup(inputGroupPos)->getMultiplier());
        }
    }
}

void SinkOperatorUtil::recomputeSchema(const Schema& inputSchema,
    const binder::expression_vector& expressionsToMerge, Schema& resultSchema) {
    DASSERT(!expressionsToMerge.empty());
    resultSchema.clear();
    mergeSchema(inputSchema, expressionsToMerge, resultSchema);
}

std::unordered_map<f_group_pos, binder::expression_vector>
SinkOperatorUtil::getUnFlatPayloadsPerGroup(const Schema& schema,
    const binder::expression_vector& payloads) {
    std::unordered_map<f_group_pos, binder::expression_vector> result;
    for (auto& payload : payloads) {
        auto groupPos = schema.getGroupPos(*payload);
        if (schema.getGroup(groupPos)->isFlat()) {
            continue;
        }
        if (!result.contains(groupPos)) {
            result.insert({groupPos, binder::expression_vector{}});
        }
        result.at(groupPos).push_back(payload);
    }
    return result;
}

binder::expression_vector SinkOperatorUtil::getFlatPayloads(const Schema& schema,
    const binder::expression_vector& payloads) {
    binder::expression_vector result;
    for (auto& payload : payloads) {
        if (schema.getGroup(payload)->isFlat()) {
            result.push_back(payload);
        }
    }
    return result;
}

uint32_t SinkOperatorUtil::appendPayloadsToNewGroup(Schema& schema,
    binder::expression_vector& payloads) {
    auto outputGroupPos = schema.createGroup();
    for (auto& payload : payloads) {
        schema.insertToGroupAndScope(payload, outputGroupPos);
    }
    return outputGroupPos;
}

} // namespace planner
} // namespace lbug