update

kuzudb · Aug 2, 2023 · cbc66db · cbc66db
1 parent 11f3390
commit cbc66db
Show file tree

Hide file tree

Showing 15 changed files with 246 additions and 171 deletions.
diff --git a/src/common/vector/auxiliary_buffer.cpp b/src/common/vector/auxiliary_buffer.cpp
@@ -33,7 +33,7 @@ list_entry_t ListAuxiliaryBuffer::addList(uint64_t listSize) {
     return listEntry;
 }
 
-void ListAuxiliaryBuffer::reserve(uint64_t numValues) {
+void ListAuxiliaryBuffer::resize(uint64_t numValues) {
     if (numValues <= capacity) {
         size = numValues;
         return;

diff --git a/src/include/common/vector/auxiliary_buffer.h b/src/include/common/vector/auxiliary_buffer.h
@@ -80,7 +80,7 @@ class ListAuxiliaryBuffer : public AuxiliaryBuffer {
 
     inline void resetSize() { size = 0; }
 
-    void reserve(uint64_t numValues);
+    void resize(uint64_t numValues);
 
 private:
     void resizeDataVector(ValueVector* dataVector);

diff --git a/src/include/common/vector/value_vector.h b/src/include/common/vector/value_vector.h
@@ -148,8 +148,8 @@ class ListVector {
         return reinterpret_cast<ListAuxiliaryBuffer*>(vector->auxiliaryBuffer.get())
             ->addList(listSize);
     }
-    static inline void reserveDataVector(ValueVector* vector, uint64_t numValues) {
-        reinterpret_cast<ListAuxiliaryBuffer*>(vector->auxiliaryBuffer.get())->reserve(numValues);
+    static inline void resizeDataVector(ValueVector* vector, uint64_t numValues) {
+        reinterpret_cast<ListAuxiliaryBuffer*>(vector->auxiliaryBuffer.get())->resize(numValues);
     }
 
     static void copyFromRowData(ValueVector* vector, uint32_t pos, const uint8_t* rowData);

diff --git a/src/include/storage/copier/column_chunk.h b/src/include/storage/copier/column_chunk.h
@@ -64,11 +64,9 @@ class ColumnChunk {
                common::BufferPoolConstants::PAGE_4KB_SIZE;
     }
 
-    inline uint64_t getNumElements() const { return numElements; }
+    virtual void writeVal(const common::Value& val, uint64_t posToWrite);
 
-    virtual void appendVal(const common::Value& val);
-
-    virtual void reserve(uint64_t numValues);
+    inline uint64_t getNumBytesPerValue() const { return numBytesPerValue; }
 
 protected:
     ColumnChunk(common::LogicalType dataType, common::offset_t numValues,
@@ -93,21 +91,19 @@ class ColumnChunk {
 
     common::offset_t getOffsetInBuffer(common::offset_t pos) const;
 
-private:
-    void reserveForBuffer(uint64_t numValues);
-
 public:
     std::unique_ptr<uint8_t[]> buffer;
 
+private:
+    virtual void resize(uint64_t numBytesToResize);
+
 protected:
     common::LogicalType dataType;
     uint32_t numBytesPerValue;
     uint64_t numBytes;
     std::unique_ptr<NullColumnChunk> nullChunk;
     std::vector<std::unique_ptr<ColumnChunk>> childrenChunks;
     const common::CopyDescription* copyDescription;
-    uint64_t numElements = 0;
-    uint64_t capacity = common::StorageConstants::NODE_GROUP_SIZE;
 };
 
 class NullColumnChunk : public ColumnChunk {
@@ -118,22 +114,12 @@ class NullColumnChunk : public ColumnChunk {
         resetNullBuffer();
     }
 
-    void reserve(uint64_t numValues) override {
-        while (capacity < numValues) {
-            capacity *= 2;
-        }
-        auto reservedNumBytes = capacity * numBytesPerValue;
-        auto reservedBuffer = std::make_unique<uint8_t[]>(reservedNumBytes);
-        memset(reservedBuffer.get(), 0 /* non null */, reservedNumBytes);
-        memcpy(reservedBuffer.get(), buffer.get(), numBytes);
-        buffer = std::move(reservedBuffer);
-        numBytes = reservedNumBytes;
-    }
-
     inline void resetNullBuffer() { memset(buffer.get(), 0 /* non null */, numBytes); }
 
     inline bool isNull(common::offset_t pos) const { return getValue<bool>(pos); }
     inline void setNull(common::offset_t pos, bool isNull) { ((bool*)buffer.get())[pos] = isNull; }
+
+    void resize(uint64_t numBytesToResize) final;
 };
 
 class FixedListColumnChunk : public ColumnChunk {
@@ -144,7 +130,7 @@ class FixedListColumnChunk : public ColumnChunk {
     void append(ColumnChunk* other, common::offset_t startPosInOtherChunk,
         common::offset_t startPosInChunk, uint32_t numValuesToAppend) final;
 
-    void appendVal(const common::Value& listVal) final;
+    void writeVal(const common::Value& fixedListVal, uint64_t posToWrite) final;
 };
 
 struct ColumnChunkFactory {

diff --git a/src/include/storage/copier/list_column_chunk.h b/src/include/storage/copier/list_column_chunk.h
@@ -12,7 +12,7 @@ class ListColumnChunk : public ColumnChunk {
 public:
     ListColumnChunk(LogicalType dataType, CopyDescription* copyDescription);
 
-    inline ColumnChunk* getDataColumnChunk() const { return dataColumnChunk.get(); }
+    inline ColumnChunk* getDataColumnChunk() const { return dataChunk.get(); }
 
     void setValueFromString(const char* value, uint64_t length, uint64_t pos);
 
@@ -25,14 +25,18 @@ class ListColumnChunk : public ColumnChunk {
     void copyListFromArrowList(
         arrow::Array* array, offset_t startPosInChunk, uint32_t numValuesToAppend);
 
-    void appendVal(const common::Value& listVal) override;
+    void writeVal(const common::Value& listVal, uint64_t posToWrite) override;
 
-    common::page_idx_t getNumPages() const final {
-        return ColumnChunk::getNumPages() + dataColumnChunk->getNumPages();
+    inline common::page_idx_t getNumPages() const final {
+        return ColumnChunk::getNumPages() + dataChunk->getNumPages();
     }
 
+    void resizeDataChunk(uint64_t numValues);
+
 private:
-    std::unique_ptr<ColumnChunk> dataColumnChunk;
+    std::unique_ptr<ColumnChunk> dataChunk;
+    uint64_t numValuesInDataChunk;
+    uint64_t capacityInDataChunk;
 };
 
 } // namespace storage

diff --git a/src/include/storage/copier/struct_column_chunk.h b/src/include/storage/copier/struct_column_chunk.h
@@ -33,7 +33,7 @@ class StructColumnChunk : public ColumnChunk {
         common::LogicalType& type, const std::string& structString);
     static std::string parseStructFieldName(const std::string& structString, uint64_t& curPos);
     std::string parseStructFieldValue(const std::string& structString, uint64_t& curPos);
-    void appendVal(const common::Value& listVal) final;
+    void writeVal(const common::Value& val, uint64_t posToWrite) final;
 };
 
 } // namespace storage

diff --git a/src/include/storage/copier/var_sized_column_chunk.h b/src/include/storage/copier/var_sized_column_chunk.h
@@ -44,7 +44,7 @@ class VarSizedColumnChunk : public ColumnChunk {
     void appendVarListColumnChunk(VarSizedColumnChunk* other, common::offset_t startPosInOtherChunk,
         common::offset_t startPosInChunk, uint32_t numValuesToAppend);
 
-    void appendVal(const common::Value& val) override;
+    void writeVal(const common::Value& val, uint64_t posToWrite) override;
 
 private:
     std::unique_ptr<InMemOverflowFile> overflowFile;

diff --git a/src/include/storage/store/list_node_column.h b/src/include/storage/store/list_node_column.h
@@ -27,16 +27,42 @@ class ListNodeColumn : public NodeColumn {
         ColumnChunk* columnChunk, common::page_idx_t startPageIdx, uint64_t nodeGroupIdx) override;
 
 private:
+    void scanUnfiltered(transaction::Transaction* transaction, common::ValueVector* nodeIDVector,
+        common::ValueVector* offsetVector, common::ValueVector* resultVector);
+
+    void scanFiltered(transaction::Transaction* transaction, common::ValueVector* nodeIDVector,
+        common::ValueVector* offsetVector, common::ValueVector* resultVector);
+
     void checkpointInMemory() override;
 
     void rollbackInMemory() override;
 
     void scanWithOffsets(transaction::Transaction* transaction,
         common::node_group_idx_t nodeGroupIdx, common::offset_t startOffsetInGroup,
         common::offset_t endOffsetInGroup, common::ValueVector* resultVector,
-        uint64_t startPosInVector) override;
+        uint64_t posToWriteListEntry) override;
+
+    common::offset_t readOffset(transaction::Transaction* transaction, common::offset_t valuePos);
 
-    common::offset_t getOffset(transaction::Transaction* transaction, common::offset_t nodeOffset);
+    void scanListOffset(transaction::Transaction* transaction, common::offset_t startNodeOffset,
+        common::node_group_idx_t nodeGroupIdx, common::ValueVector* offsetVector);
+
+    inline common::offset_t readListOffsetInStorage(
+        transaction::Transaction* transaction, common::offset_t nodeOffset) {
+        return nodeOffset == 0 ? 0 : readOffset(transaction, nodeOffset - 1);
+    }
+
+    inline common::offset_t getListOffsetInStorage(transaction::Transaction* transaction,
+        common::ValueVector* offsetVector, common::offset_t nodeOffset, uint64_t nodePos) {
+        return nodePos == 0 ? readListOffsetInStorage(transaction, nodeOffset) :
+                              offsetVector->getValue<common::offset_t>(nodePos - 1);
+    }
+
+    inline uint64_t getListLength(transaction::Transaction* transaction,
+        common::ValueVector* offsetVector, common::offset_t nodeOffset, uint64_t nodePos) {
+        return getListOffsetInStorage(transaction, offsetVector, nodeOffset + 1, nodePos + 1) -
+               getListOffsetInStorage(transaction, offsetVector, nodeOffset, nodePos);
+    }
 
 private:
     std::unique_ptr<NodeColumn> dataNodeColumn;

diff --git a/src/storage/copier/column_chunk.cpp b/src/storage/copier/column_chunk.cpp
@@ -50,26 +50,6 @@ void ColumnChunk::append(ColumnChunk* other, common::offset_t startPosInOtherChu
         numValuesToAppend * numBytesPerValue);
 }
 
-void ColumnChunk::reserve(uint64_t numValues) {
-    if (numValues <= capacity) {
-        return;
-    }
-    while (capacity < numValues) {
-        capacity *= 2;
-    }
-    auto reservedNumBytes = capacity * numBytesPerValue;
-    auto reservedBuffer = std::make_unique<uint8_t[]>(reservedNumBytes);
-    memcpy(reservedBuffer.get(), buffer.get(), numBytes);
-    numBytes = reservedNumBytes;
-    buffer = std::move(reservedBuffer);
-    if (nullChunk) {
-        nullChunk->reserve(numValues);
-    }
-    for (auto& child : childrenChunks) {
-        child->reserve(numValues);
-    }
-}
-
 void ColumnChunk::append(
     arrow::Array* array, offset_t startPosInChunk, uint32_t numValuesToAppend) {
     switch (array->type_id()) {
@@ -126,35 +106,35 @@ void ColumnChunk::append(
     }
 }
 
-void ColumnChunk::appendVal(const common::Value& val) {
-    nullChunk->setNull(numElements, val.isNull());
+void ColumnChunk::writeVal(const common::Value& val, uint64_t posToWrite) {
+    nullChunk->setNull(posToWrite, val.isNull());
     switch (dataType.getPhysicalType()) {
     case PhysicalTypeID::BOOL: {
-        setValue(val.getValue<bool>(), numElements++);
+        setValue(val.getValue<bool>(), posToWrite);
     } break;
     case PhysicalTypeID::INT64: {
-        setValue(val.getValue<int64_t>(), numElements++);
+        setValue(val.getValue<int64_t>(), posToWrite);
     } break;
     case PhysicalTypeID::INT32: {
-        setValue(val.getValue<int32_t>(), numElements++);
+        setValue(val.getValue<int32_t>(), posToWrite);
     } break;
     case PhysicalTypeID::INT16: {
-        setValue(val.getValue<int16_t>(), numElements++);
+        setValue(val.getValue<int16_t>(), posToWrite);
     } break;
     case PhysicalTypeID::DOUBLE: {
-        setValue(val.getValue<double_t>(), numElements++);
+        setValue(val.getValue<double_t>(), posToWrite);
     } break;
     case PhysicalTypeID::FLOAT: {
-        setValue(val.getValue<float_t>(), numElements++);
+        setValue(val.getValue<float_t>(), posToWrite);
     } break;
     case PhysicalTypeID::INTERVAL: {
-        setValue(val.getValue<interval_t>(), numElements++);
+        setValue(val.getValue<interval_t>(), posToWrite);
     } break;
     case PhysicalTypeID::INTERNAL_ID: {
-        setValue(val.getValue<internalID_t>(), numElements++);
+        setValue(val.getValue<internalID_t>(), posToWrite);
     } break;
     default: {
-        throw NotImplementedException{"ColumnChunk::appendVal"};
+        throw NotImplementedException{"ColumnChunk::writeVal"};
     }
     }
 }
@@ -307,11 +287,34 @@ void FixedListColumnChunk::append(ColumnChunk* other, common::offset_t startPosI
     }
 }
 
-void FixedListColumnChunk::appendVal(const common::Value& listVal) {
-    assert(listVal.getDataType()->getPhysicalType() == PhysicalTypeID::FIXED_LIST);
-    auto numElements = NestedVal::getChildrenSize(&listVal);
-    for (auto i = 0u; i < numElements; i++) {
-        throw NotImplementedException{"here"};
+void FixedListColumnChunk::writeVal(const common::Value& fixedListVal, uint64_t posToWrite) {
+    assert(fixedListVal.getDataType()->getPhysicalType() == PhysicalTypeID::FIXED_LIST);
+    auto numValues = NestedVal::getChildrenSize(&fixedListVal);
+    auto numBytesPerValueInList = getDataTypeSizeInChunk(*fixedListVal.getDataType());
+    auto bufferToWrite = buffer.get() + posToWrite * numBytesPerValue;
+    for (auto i = 0u; i < numValues; i++) {
+        auto val = common::NestedVal::getChildVal(&fixedListVal, i);
+        switch (fixedListVal.getDataType()->getPhysicalType()) {
+        case PhysicalTypeID::INT64: {
+            memcpy(bufferToWrite, &val->getValueReference<int64_t>(), numBytesPerValueInList);
+        } break;
+        case PhysicalTypeID::INT32: {
+            memcpy(bufferToWrite, &val->getValueReference<int32_t>(), numBytesPerValueInList);
+        } break;
+        case PhysicalTypeID::INT16: {
+            memcpy(bufferToWrite, &val->getValueReference<int16_t>(), numBytesPerValueInList);
+        } break;
+        case PhysicalTypeID::DOUBLE: {
+            memcpy(bufferToWrite, &val->getValueReference<double_t>(), numBytesPerValueInList);
+        } break;
+        case PhysicalTypeID::FLOAT: {
+            memcpy(bufferToWrite, &val->getValueReference<float_t>(), numBytesPerValueInList);
+        } break;
+        default: {
+            throw NotImplementedException{"FixedListColumnChunk::writeVal"};
+        }
+        }
+        bufferToWrite += numBytesPerValueInList;
     }
 }
 
@@ -394,5 +397,26 @@ common::offset_t ColumnChunk::getOffsetInBuffer(common::offset_t pos) const {
     return offsetInBuffer;
 }
 
+void ColumnChunk::resize(uint64_t numBytesToResize) {
+    auto reservedBuffer = std::make_unique<uint8_t[]>(numBytesToResize);
+    memcpy(reservedBuffer.get(), buffer.get(), numBytes);
+    numBytes = numBytesToResize;
+    buffer = std::move(reservedBuffer);
+    if (nullChunk) {
+        nullChunk->resize(numBytesToResize);
+    }
+    for (auto& child : childrenChunks) {
+        child->resize(numBytesToResize);
+    }
+}
+
+void NullColumnChunk::resize(uint64_t numBytesToResize) {
+    auto reservedBuffer = std::make_unique<uint8_t[]>(numBytesToResize);
+    memset(reservedBuffer.get(), 0 /* non null */, numBytesToResize);
+    memcpy(reservedBuffer.get(), buffer.get(), numBytes);
+    buffer = std::move(reservedBuffer);
+    numBytes = numBytesToResize;
+}
+
 } // namespace storage
 } // namespace kuzu