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

#include "processor/operator/unwind_dedup.h"

#include "common/vector/value_vector.h"
#include "function/hash/vector_hash_functions.h"
#include "processor/execution_context.h"
#include "storage/buffer_manager/memory_manager.h"

using namespace lbug::common;
using namespace lbug::function;
using namespace lbug::storage;

namespace lbug {
namespace processor {

void UnwindDedup::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* /*context*/) {
    keyVectors.clear();
    keyVectors.reserve(keyDataPositions.size());
    for (auto& keyDataPos : keyDataPositions) {
        keyVectors.push_back(resultSet->getValueVector(keyDataPos).get());
    }
}

bool UnwindDedup::getNextTuplesInternal(ExecutionContext* context) {
    while (true) {
        if (!children[0]->getNextTuple(context)) {
            return false;
        }

        // Compute hash for the key vector
        DASSERT(!keyVectors.empty());
        const auto& selVector = keyVectors[0]->state->getSelVector();
        if (selVector.getSelSize() == 0) {
            continue;
        }

        auto hashVector = std::make_unique<ValueVector>(LogicalType::HASH(),
            MemoryManager::Get(*context->clientContext));
        hashVector->state = keyVectors[0]->state;
        VectorHashFunction::computeHash(*keyVectors[0], selVector, *hashVector, selVector);
        if (keyVectors.size() > 1) {
            auto tmpHashVector = std::make_unique<ValueVector>(LogicalType::HASH(),
                MemoryManager::Get(*context->clientContext));
            tmpHashVector->state = keyVectors[0]->state;
            for (auto i = 1u; i < keyVectors.size(); ++i) {
                DASSERT(keyVectors[i]->state == keyVectors[0]->state);
                VectorHashFunction::computeHash(*keyVectors[i], selVector, *tmpHashVector,
                    selVector);
                VectorHashFunction::combineHash(*hashVector, selVector, *tmpHashVector, selVector,
                    *hashVector, selVector);
            }
        }

        // Filter out duplicate values by modifying the selection vector
        auto hashData = reinterpret_cast<hash_t*>(hashVector->getData());
        sel_t newSelectedPositions[DEFAULT_VECTOR_CAPACITY];
        auto newSelSize = 0u;

        for (auto i = 0u; i < selVector.getSelSize(); i++) {
            auto pos = selVector[i];
            auto hash = hashData[pos];

            if (seenHashes.find(hash) == seenHashes.end()) {
                seenHashes.insert(hash);
                newSelectedPositions[newSelSize++] = pos;
            }
        }

        if (newSelSize > 0) {
            // Update the selection vector to only include non-duplicate values
            auto& selVectorUnsafe = keyVectors[0]->state->getSelVectorUnsafe();
            selVectorUnsafe.setToFiltered(newSelSize);
            auto buffer = selVectorUnsafe.getMutableBuffer();
            for (auto i = 0u; i < newSelSize; i++) {
                buffer[i] = newSelectedPositions[i];
            }
            return true;
        }
        // If all values were duplicates, continue to next batch
    }
}

} // namespace processor
} // namespace lbug