lbug 0.17.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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

#include "optimizer/unwind_dedup_optimizer.h"

#include <unordered_set>

#include "binder/expression/rel_expression.h"
#include "planner/operator/logical_hash_join.h"
#include "planner/operator/logical_unwind.h"
#include "planner/operator/logical_unwind_deduplicate.h"
#include "planner/operator/persistent/logical_merge.h"

using namespace lbug::common;
using namespace lbug::planner;

namespace lbug {
namespace optimizer {

void UnwindDedupOptimizer::rewrite(LogicalPlan* plan) {
    visitOperator(plan->getLastOperator(), true /* isRoot */);
}

std::shared_ptr<LogicalOperator> UnwindDedupOptimizer::visitOperator(
    const std::shared_ptr<LogicalOperator>& op, bool isRoot) {
    // bottom-up traversal
    for (auto i = 0u; i < op->getNumChildren(); ++i) {
        op->setChild(i, visitOperator(op->getChild(i), false /* isRoot */));
    }
    auto canRewriteParentMerge = canRewriteCurrentMerge;
    canRewriteCurrentMerge = isRoot;
    auto result = visitOperatorReplaceSwitch(op);
    canRewriteCurrentMerge = canRewriteParentMerge;
    result->computeFlatSchema();
    return result;
}

// Helper function to recursively find UNWIND in the operator tree
static std::shared_ptr<LogicalUnwind> findUnwind(std::shared_ptr<LogicalOperator> op) {
    if (op->getOperatorType() == LogicalOperatorType::UNWIND) {
        return std::static_pointer_cast<LogicalUnwind>(op);
    }
    // Recursively search through children
    for (auto i = 0u; i < op->getNumChildren(); ++i) {
        auto result = findUnwind(op->getChild(i));
        if (result != nullptr) {
            return result;
        }
    }
    return nullptr;
}

static bool appendKeyIfInScope(binder::expression_vector& keys,
    std::unordered_set<std::string>& names, const LogicalOperator& op,
    const std::shared_ptr<binder::Expression>& key) {
    auto schema = op.getSchema();
    if (!schema->isExpressionInScope(*key)) {
        return false;
    }
    if (names.insert(key->getUniqueName()).second) {
        keys.push_back(key);
    }
    return true;
}

static binder::expression_vector getDedupKeyExpressions(const LogicalMerge& merge,
    const LogicalOperator& op) {
    binder::expression_vector result;
    std::unordered_set<std::string> keyNames;
    for (auto& key : merge.getKeys()) {
        if (!appendKeyIfInScope(result, keyNames, op, key)) {
            return {};
        }
    }
    for (auto& info : merge.getInsertRelInfos()) {
        auto rel = info.pattern->constPtrCast<binder::RelExpression>();
        if (!appendKeyIfInScope(result, keyNames, op, rel->getSrcNode()->getInternalID())) {
            return {};
        }
        if (!appendKeyIfInScope(result, keyNames, op, rel->getDstNode()->getInternalID())) {
            return {};
        }
    }
    return result;
}

std::shared_ptr<LogicalOperator> UnwindDedupOptimizer::visitMergeReplace(
    std::shared_ptr<LogicalOperator> op) {
    auto merge = op->ptrCast<LogicalMerge>();
    if (merge == nullptr) {
        return op;
    }
    if (!canRewriteCurrentMerge && !merge->getInsertRelInfos().empty()) {
        return op;
    }

    // Check if MERGE has ON MATCH or ON CREATE clauses
    // If it does, we should NOT apply UNWIND_DEDUP because duplicates need different handling:
    // - First occurrence: ON CREATE
    // - Subsequent occurrences: ON MATCH
    bool hasOnMatchOrOnCreate =
        !merge->getOnMatchSetNodeInfos().empty() || !merge->getOnMatchSetRelInfos().empty() ||
        !merge->getOnCreateSetNodeInfos().empty() || !merge->getOnCreateSetRelInfos().empty();
    if (hasOnMatchOrOnCreate) {
        return op;
    }

    auto mergeChild = merge->getChild(0);
    if (mergeChild->getOperatorType() != LogicalOperatorType::HASH_JOIN) {
        return op;
    }
    auto hashJoin = mergeChild->ptrCast<LogicalHashJoin>();
    auto probeChild = hashJoin->getChild(0);

    // Try to find UNWIND either as direct child or nested deeper
    std::shared_ptr<LogicalUnwind> unwind;
    if (probeChild->getOperatorType() == LogicalOperatorType::UNWIND) {
        // Direct UNWIND child (original case)
        unwind = std::static_pointer_cast<LogicalUnwind>(probeChild);
    } else {
        // Search recursively for UNWIND (handles FLATTEN and other intermediate operators)
        unwind = findUnwind(probeChild);
    }

    if (unwind != nullptr) {
        // Place UNWIND_DEDUP after the join so MERGE keys that depend on matched nodes are in
        // scope.
        auto keyExpressions = getDedupKeyExpressions(*merge, *mergeChild);
        if (keyExpressions.empty()) {
            return op;
        }
        auto dedup =
            std::make_shared<LogicalUnwindDeduplicate>(mergeChild, std::move(keyExpressions));
        dedup->computeFlatSchema();
        merge->setChild(0, dedup);
        return op;
    }

    return op;
}

} // namespace optimizer
} // namespace lbug