lbug 0.22.0

An in-process property graph database management system built for query speed and scalability
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

#include "binder/expression/property_expression.h"
#include "catalog/catalog_entry/node_table_catalog_entry.h"
#include "main/client_context.h"
#include "planner/operator/scan/logical_grouped_reachable_count.h"
#include "processor/operator/scan/grouped_reachable_count.h"
#include "processor/plan_mapper.h"
#include "storage/storage_manager.h"

using namespace lbug::binder;
using namespace lbug::catalog;
using namespace lbug::common;
using namespace lbug::graph;
using namespace lbug::planner;

namespace lbug {
namespace processor {

std::unique_ptr<PhysicalOperator> PlanMapper::mapGroupedReachableCount(
    const LogicalOperator* logicalOperator) {
    auto& logical = logicalOperator->constCast<LogicalGroupedReachableCount>();
    auto outSchema = logical.getSchema();

    auto boundNode = logical.getBoundNode();
    auto nbrNode = logical.getNbrNode();
    DASSERT(boundNode->getNumEntries() == 1 && nbrNode->getNumEntries() == 1);
    auto boundTableID = boundNode->getTableIDs()[0];
    auto nbrTableID = nbrNode->getTableIDs()[0];

    std::vector<TableCatalogEntry*> nodeEntries{boundNode->getEntry(0), nbrNode->getEntry(0)};
    std::vector<TableCatalogEntry*> relEntries{logical.getRelGroupEntry()};
    auto graphEntry = NativeGraphEntry(std::move(nodeEntries), std::move(relEntries));

    DASSERT(logical.getDirection() == ExtendDirection::FWD);
    RelDataDirection direction = RelDataDirection::FWD;

    auto* storageManager = storage::StorageManager::Get(*clientContext);
    auto* nodeTable = storageManager->getTable(nbrTableID)->ptrCast<storage::NodeTable>();
    auto* nodeEntry = nbrNode->getEntry(0)->ptrCast<catalog::NodeTableCatalogEntry>();

    std::vector<GroupedReachableCount::KeyInfo> keyInfos;
    keyInfos.reserve(logical.getKeys().size());
    std::vector<DataPos> keyOutputPoss;
    keyOutputPoss.reserve(logical.getKeys().size());
    for (auto& key : logical.getKeys()) {
        auto& prop = key->constCast<PropertyExpression>();
        GroupedReachableCount::KeyInfo info;
        info.logicalType = key->getDataType().copy();
        if (prop.isInternalID()) {
            info.isInternalID = true;
        } else {
            info.columnID = nodeEntry->getColumnID(prop.getPropertyName());
        }
        keyInfos.push_back(std::move(info));
        keyOutputPoss.push_back(getDataPos(*key, *outSchema));
    }

    column_id_t scoreColumnID = INVALID_COLUMN_ID;
    LogicalType scoreType;
    if (logical.hasAvg()) {
        auto& prop = logical.getAvgChild()->constCast<PropertyExpression>();
        scoreColumnID = nodeEntry->getColumnID(prop.getPropertyName());
        scoreType = logical.getAvgChild()->getDataType().copy();
    }

    DataPos countOutputPos;
    bool hasCount = logical.hasCount();
    if (hasCount) {
        countOutputPos = getDataPos(*logical.getCountExpr(), *outSchema);
    }
    DataPos avgOutputPos;
    bool hasAvg = logical.hasAvg();
    if (hasAvg) {
        avgOutputPos = getDataPos(*logical.getAvgExpr(), *outSchema);
    }
    auto printInfo = std::make_unique<GroupedReachableCountPrintInfo>(
        logical.getRelGroupEntry()->getName(), logical.getLowerBound(), logical.getUpperBound(),
        logical.getKeys().size(), hasCount, hasAvg);
    return std::make_unique<GroupedReachableCount>(std::move(graphEntry), direction,
        logical.getLowerBound(), logical.getUpperBound(), boundTableID, nbrTableID, nodeTable,
        std::move(keyInfos), scoreColumnID, std::move(scoreType), std::move(keyOutputPoss),
        std::move(countOutputPos), hasCount, std::move(avgOutputPos), hasAvg, getOperatorID(),
        std::move(printInfo));
}

} // namespace processor
} // namespace lbug