-
Notifications
You must be signed in to change notification settings - Fork 86
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
23 changed files
with
844 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
#pragma once | ||
|
||
#include <queue> | ||
|
||
#include "common/types/types_include.h" | ||
#include "database.h" | ||
|
||
namespace kuzu { | ||
namespace storage { | ||
class Column; | ||
} | ||
|
||
namespace main { | ||
|
||
//// Note: pooling is tricky to figure out "Busy state" | ||
//class ThreadPool { | ||
//public: | ||
// ThreadPool(size_t numThreads); | ||
// ~ThreadPool(); | ||
// | ||
// template<typename F, typename... Args> | ||
// void enqueue(F&& f, Args&&... args) { | ||
// { | ||
// std::unique_lock<std::mutex> lock(queueMtx); | ||
// tasks.emplace([=] { std::invoke(f, args...); }); | ||
// } | ||
// condition.notify_one(); | ||
// } | ||
// | ||
//private: | ||
// std::vector<std::thread> threads; | ||
// std::queue<std::function<void()>> tasks; | ||
// std::mutex queueMtx; | ||
// std::condition_variable condition; | ||
// bool stop = false; | ||
//}; | ||
|
||
class StorageDriver { | ||
public: | ||
explicit StorageDriver(Database* database, size_t numThreads = 1); | ||
|
||
~StorageDriver(); | ||
|
||
std::pair<std::unique_ptr<uint8_t[]>, size_t> scan(const std::string& nodeName, | ||
const std::string& propertyName, common::offset_t* offsets, size_t size); | ||
|
||
private: | ||
void scanColumn(storage::Column* column, common::offset_t* offsets, size_t size, uint8_t* result); | ||
|
||
private: | ||
catalog::Catalog* catalog; | ||
storage::StorageManager* storageManager; | ||
size_t numThreads; | ||
}; | ||
|
||
} // namespace main | ||
} // namespace kuzu |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,82 @@ | ||
#include "main/storage_driver.h" | ||
|
||
#include "catalog/catalog.h" | ||
#include "storage/storage_manager.h" | ||
|
||
using namespace kuzu::common; | ||
|
||
namespace kuzu { | ||
namespace main { | ||
|
||
// ThreadPool::ThreadPool(size_t numThreads) { | ||
// for (auto i = 0u; i < numThreads; ++i) { | ||
// threads.emplace_back([this] { | ||
// while (true) { | ||
// std::function<void()> task; | ||
// { | ||
// std::unique_lock<std::mutex> lck(queueMtx); | ||
// condition.wait(lck, [this] { return stop || !tasks.empty(); }); | ||
// if (stop && tasks.empty()) { | ||
// return; | ||
// } | ||
// task = std::move(tasks.front()); | ||
// tasks.pop(); | ||
// } | ||
// task(); | ||
// } | ||
// }); | ||
// } | ||
//} | ||
// | ||
// ThreadPool::~ThreadPool() { | ||
// { | ||
// std::unique_lock<std::mutex> lck(queueMtx); | ||
// stop = true; | ||
// } | ||
// condition.notify_all(); | ||
// for (auto& thread : threads) { | ||
// thread.join(); | ||
// } | ||
//} | ||
|
||
StorageDriver::StorageDriver(kuzu::main::Database* database, size_t numThreads) | ||
: catalog{database->catalog.get()}, storageManager{database->storageManager.get()}, | ||
numThreads{numThreads} {} | ||
|
||
StorageDriver::~StorageDriver() = default; | ||
|
||
std::pair<std::unique_ptr<uint8_t[]>, size_t> StorageDriver::scan(const std::string& nodeName, | ||
const std::string& propertyName, common::offset_t* offsets, size_t size) { | ||
// Resolve files to read from | ||
auto catalogContent = catalog->getReadOnlyVersion(); | ||
auto nodeTableID = catalogContent->getTableID(nodeName); | ||
auto propertyID = catalogContent->getTableSchema(nodeTableID)->getPropertyID(propertyName); | ||
auto nodeTable = storageManager->getNodesStore().getNodeTable(nodeTableID); | ||
auto column = nodeTable->getPropertyColumn(propertyID); | ||
|
||
auto bufferSize = column->elementSize * size; | ||
auto result = std::make_unique<uint8_t[]>(bufferSize); | ||
auto buffer = result.get(); | ||
std::vector<std::thread> threads; | ||
auto numElementsPerThread = size / numThreads + 1; | ||
auto sizeLeft = size; | ||
while (sizeLeft > 0) { | ||
auto sizeToRead = std::min(numElementsPerThread, sizeLeft); | ||
threads.emplace_back(&StorageDriver::scanColumn, this, column, offsets, sizeToRead, buffer); | ||
offsets += sizeToRead; | ||
buffer += sizeToRead * column->elementSize; | ||
sizeLeft -= sizeToRead; | ||
} | ||
for (auto& thread : threads) { | ||
thread.join(); | ||
} | ||
return std::make_pair(std::move(result), bufferSize); | ||
} | ||
|
||
void StorageDriver::scanColumn( | ||
storage::Column* column, common::offset_t* offsets, size_t size, uint8_t* result) { | ||
column->scan(offsets, size, result); | ||
} | ||
|
||
} // namespace main | ||
} // namespace kuzu |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,26 @@ | ||
#include "main/storage_driver.h" | ||
#include "main_test_helper/main_test_helper.h" | ||
|
||
using namespace kuzu::testing; | ||
using namespace kuzu::common; | ||
|
||
TEST_F(ApiTest, StorageDriverScan) { | ||
auto storageDriver = std::make_unique<StorageDriver>(database.get(), 3); | ||
auto size = 6; | ||
auto nodeOffsetsBuffer = std::make_unique<uint8_t[]>(sizeof(offset_t) * size); | ||
auto nodeOffsets = (offset_t*)nodeOffsetsBuffer.get(); | ||
nodeOffsets[0] = 7; | ||
nodeOffsets[1] = 0; | ||
nodeOffsets[2] = 3; | ||
nodeOffsets[3] = 1; | ||
nodeOffsets[4] = 2; | ||
nodeOffsets[5] = 6; | ||
auto result = storageDriver->scan("person", "ID", nodeOffsets, size); | ||
auto ids = (int64_t*)result.first.get(); | ||
ASSERT_EQ(ids[0], 10); | ||
ASSERT_EQ(ids[1], 0); | ||
ASSERT_EQ(ids[2], 5); | ||
ASSERT_EQ(ids[3], 2); | ||
ASSERT_EQ(ids[4], 3); | ||
ASSERT_EQ(ids[5], 9); | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,5 +1,3 @@ | ||
from ._kuzu import * | ||
# The following imports will override C++ implementations with Python | ||
# implementations. | ||
from .database import * | ||
from .connection import * | ||
from .query_result import * |
Oops, something went wrong.