Merge pull request #2001 from kuzudb/fix-lookup

Fix issue 1915
kuzudb · Sep 7, 2023 · d3ed7f2 · d3ed7f2
2 parents ad5fa79 + 3610025
commit d3ed7f2
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 26 deletions.
diff --git a/src/include/storage/store/node_column.h b/src/include/storage/store/node_column.h
@@ -124,6 +124,8 @@ class NodeColumn {
     virtual void writeValue(common::offset_t nodeOffset, common::ValueVector* vectorToWriteFrom,
         uint32_t posInVectorToWriteFrom);
 
+    PageElementCursor getPageCursorForOffset(
+        transaction::TransactionType transactionType, common::offset_t nodeOffset);
     // TODO(Guodong): This is mostly duplicated with
     // StorageStructure::createWALVersionOfPageIfNecessaryForElement(). Should be cleared later.
     WALPageIdxPosInPageAndFrame createWALVersionOfPageForValue(common::offset_t nodeOffset);

diff --git a/src/storage/store/node_column.cpp b/src/storage/store/node_column.cpp
@@ -122,9 +122,7 @@ void NodeColumn::batchLookup(
     Transaction* transaction, const offset_t* nodeOffsets, size_t size, uint8_t* result) {
     for (auto i = 0u; i < size; ++i) {
         auto nodeOffset = nodeOffsets[i];
-        auto nodeGroupIdx = StorageUtils::getNodeGroupIdx(nodeOffset);
-        auto cursor = PageUtils::getPageElementCursorForPos(nodeOffset, numValuesPerPage);
-        cursor.pageIdx += metadataDA->get(nodeGroupIdx, transaction->getType()).pageIdx;
+        auto cursor = getPageCursorForOffset(transaction->getType(), nodeOffset);
         readFromPage(transaction, cursor.pageIdx, [&](uint8_t* frame) -> void {
             memcpy(result + i * numBytesPerFixedSizedValue,
                 frame + (cursor.elemPosInPage * numBytesPerFixedSizedValue),
@@ -137,9 +135,7 @@ void BoolNodeColumn::batchLookup(
     Transaction* transaction, const offset_t* nodeOffsets, size_t size, uint8_t* result) {
     for (auto i = 0u; i < size; ++i) {
         auto nodeOffset = nodeOffsets[i];
-        auto nodeGroupIdx = StorageUtils::getNodeGroupIdx(nodeOffset);
-        auto cursor = PageUtils::getPageElementCursorForPos(nodeOffset, numValuesPerPage);
-        cursor.pageIdx += metadataDA->get(nodeGroupIdx, transaction->getType()).pageIdx;
+        auto cursor = getPageCursorForOffset(transaction->getType(), nodeOffset);
         readFromPage(transaction, cursor.pageIdx, [&](uint8_t* frame) -> void {
             // De-compress bitpacked bools
             result[i] = NullMask::isNull((uint64_t*)frame, cursor.elemPosInPage);
@@ -182,17 +178,12 @@ void NodeColumn::scanInternal(
     Transaction* transaction, ValueVector* nodeIDVector, ValueVector* resultVector) {
     auto startNodeOffset = nodeIDVector->readNodeOffset(0);
     assert(startNodeOffset % DEFAULT_VECTOR_CAPACITY == 0);
-    auto nodeGroupIdx = StorageUtils::getNodeGroupIdx(startNodeOffset);
-    auto offsetInNodeGroup =
-        startNodeOffset - StorageUtils::getStartOffsetOfNodeGroup(nodeGroupIdx);
-    auto pageCursor = PageUtils::getPageElementCursorForPos(offsetInNodeGroup, numValuesPerPage);
-    auto chunkMeta = metadataDA->get(nodeGroupIdx, transaction->getType());
-    pageCursor.pageIdx += chunkMeta.pageIdx;
+    auto cursor = getPageCursorForOffset(transaction->getType(), startNodeOffset);
     if (nodeIDVector->state->selVector->isUnfiltered()) {
         scanUnfiltered(
-            transaction, pageCursor, nodeIDVector->state->selVector->selectedSize, resultVector);
+            transaction, cursor, nodeIDVector->state->selVector->selectedSize, resultVector);
     } else {
-        scanFiltered(transaction, pageCursor, nodeIDVector, resultVector);
+        scanFiltered(transaction, cursor, nodeIDVector, resultVector);
     }
 }
 
@@ -259,11 +250,9 @@ void NodeColumn::lookupInternal(
 
 void NodeColumn::lookupValue(transaction::Transaction* transaction, offset_t nodeOffset,
     ValueVector* resultVector, uint32_t posInVector) {
-    auto nodeGroupIdx = StorageUtils::getNodeGroupIdx(nodeOffset);
-    auto pageCursor = PageUtils::getPageElementCursorForPos(nodeOffset, numValuesPerPage);
-    pageCursor.pageIdx += metadataDA->get(nodeGroupIdx, transaction->getType()).pageIdx;
-    readFromPage(transaction, pageCursor.pageIdx, [&](uint8_t* frame) -> void {
-        readNodeColumnFunc(frame, pageCursor, resultVector, posInVector, 1 /* numValuesToRead */);
+    auto cursor = getPageCursorForOffset(transaction->getType(), nodeOffset);
+    readFromPage(transaction, cursor.pageIdx, [&](uint8_t* frame) -> void {
+        readNodeColumnFunc(frame, cursor, resultVector, posInVector, 1 /* numValuesToRead */);
     });
 }
 
@@ -354,10 +343,7 @@ void NodeColumn::writeValue(
 }
 
 WALPageIdxPosInPageAndFrame NodeColumn::createWALVersionOfPageForValue(offset_t nodeOffset) {
-    auto nodeGroupIdx = StorageUtils::getNodeGroupIdx(nodeOffset);
-    auto originalPageCursor = PageUtils::getPageElementCursorForPos(
-        nodeOffset - StorageUtils::getStartOffsetOfNodeGroup(nodeGroupIdx), numValuesPerPage);
-    originalPageCursor.pageIdx += metadataDA->get(nodeGroupIdx, TransactionType::WRITE).pageIdx;
+    auto originalPageCursor = getPageCursorForOffset(TransactionType::WRITE, nodeOffset);
     bool insertingNewPage = false;
     if (originalPageCursor.pageIdx >= dataFH->getNumPages()) {
         assert(originalPageCursor.pageIdx == dataFH->getNumPages());
@@ -396,9 +382,8 @@ void NodeColumn::rollbackInMemory() {
     }
 }
 
-void NodeColumn::populateWithDefaultVal(const catalog::Property& property,
-    kuzu::storage::NodeColumn* nodeColumn, common::ValueVector* defaultValueVector,
-    uint64_t numNodeGroups) {
+void NodeColumn::populateWithDefaultVal(const Property& property, NodeColumn* nodeColumn,
+    ValueVector* defaultValueVector, uint64_t numNodeGroups) {
     auto columnChunk = ColumnChunkFactory::createColumnChunk(
         *property.getDataType(), nullptr /* copyDescription */);
     columnChunk->populateWithDefaultVal(defaultValueVector);
@@ -409,6 +394,15 @@ void NodeColumn::populateWithDefaultVal(const catalog::Property& property,
     }
 }
 
+PageElementCursor NodeColumn::getPageCursorForOffset(
+    TransactionType transactionType, offset_t nodeOffset) {
+    auto nodeGroupIdx = StorageUtils::getNodeGroupIdx(nodeOffset);
+    auto offsetInNodeGroup = nodeOffset - StorageUtils::getStartOffsetOfNodeGroup(nodeGroupIdx);
+    auto pageCursor = PageUtils::getPageElementCursorForPos(offsetInNodeGroup, numValuesPerPage);
+    pageCursor.pageIdx += metadataDA->get(nodeGroupIdx, transactionType).pageIdx;
+    return pageCursor;
+}
+
 // Page size must be aligned to 8 byte chunks for the 64-bit NullMask algorithms to work
 // without the possibility of memory errors from reading/writing off the end of a page.
 static_assert(PageUtils::getNumElementsInAPage(1, false /*requireNullColumn*/) % 8 == 0);