in place commit

kuzudb · Mar 20, 2024 · 65ccd6f · 65ccd6f
1 parent ebe73f5
commit 65ccd6f
Show file tree

Hide file tree

Showing 5 changed files with 222 additions and 376 deletions.
diff --git a/src/include/storage/store/var_list_column.h b/src/include/storage/store/var_list_column.h
@@ -42,6 +42,28 @@ struct ListOffsetSizeInfo {
     bool isOffsetSortedAscending(uint64_t startPos, uint64_t endPos) const;
 };
 
+struct VarListInfo {
+    common::offset_t totalListNum;
+    std::unique_ptr<ChunkCollection> dataInsertUpdateChunkCollection;
+    std::unique_ptr<ChunkCollection> sizeUpdateChunkCollection;
+    std::unique_ptr<ChunkCollection> sizeInsertChunkCollection;
+    std::unique_ptr<ChunkCollection> offsetUpdateChunkCollection;
+    std::unique_ptr<ChunkCollection> offsetInsertChunkCollection;
+
+    VarListInfo(common::offset_t totalListNum,
+        std::unique_ptr<ChunkCollection> dataInsertUpdateChunkCollection,
+        std::unique_ptr<ChunkCollection> sizeUpdateChunkCollection,
+        std::unique_ptr<ChunkCollection> sizeInsertChunkCollection,
+        std::unique_ptr<ChunkCollection> offsetUpdateChunkCollection,
+        std::unique_ptr<ChunkCollection> offsetInsertChunkCollection)
+        : totalListNum{totalListNum},
+          dataInsertUpdateChunkCollection{std::move(dataInsertUpdateChunkCollection)},
+          sizeUpdateChunkCollection{std::move(sizeUpdateChunkCollection)},
+          sizeInsertChunkCollection{std::move(sizeInsertChunkCollection)},
+          offsetUpdateChunkCollection{std::move(offsetUpdateChunkCollection)},
+          offsetInsertChunkCollection{std::move(offsetInsertChunkCollection)} {}
+};
+
 class VarListColumn : public Column {
     friend class VarListLocalColumn;
 
@@ -77,40 +99,22 @@ class VarListColumn : public Column {
     void scanFiltered(transaction::Transaction* transaction, common::node_group_idx_t nodeGroupIdx,
         common::ValueVector* offsetVector, const ListOffsetSizeInfo& listOffsetInfoInStorage);
 
-    void write(common::node_group_idx_t nodeGroupIdx, common::offset_t offsetInChunk,
-        common::ValueVector* vectorToWriteFrom, uint32_t posInVectorToWriteFrom) override;
-
-    void write(common::node_group_idx_t nodeGroupIdx, common::offset_t offsetInChunk,
-        ColumnChunk* data, common::offset_t dataOffset, common::length_t numValues) override;
-
-    bool canCommitInPlace(transaction::Transaction* transaction,
-        common::node_group_idx_t nodeGroupIdx, const ChunkCollection& localInsertChunks,
-        const offset_to_row_idx_t& insertInfo, const ChunkCollection& localUpdateChunks,
-        const offset_to_row_idx_t& updateInfo) override;
-    bool canCommitInPlace(transaction::Transaction* transaction,
-        common::node_group_idx_t nodeGroupIdx, const std::vector<common::offset_t>& dstOffsets,
-        ColumnChunk* chunk, common::offset_t srcOffset) override;
-
-    void commitLocalChunkInPlace(transaction::Transaction* transaction,
-        common::node_group_idx_t nodeGroupIdx, const ChunkCollection& localInsertChunks,
-        const offset_to_row_idx_t& insertInfo, const ChunkCollection& localUpdateChunks,
-        const offset_to_row_idx_t& updateInfo, const offset_set_t& deleteInfo) override;
-
-    void commitColumnChunkInPlace(common::node_group_idx_t nodeGroupIdx,
-        const std::vector<common::offset_t>& dstOffsets, ColumnChunk* chunk,
-        common::offset_t srcOffset) override;
-
-    std::pair<std::unique_ptr<ChunkCollection>,std::unique_ptr<ChunkCollection>> extractDataChunks(const ChunkCollection& localInsertChunks,
+    void commitOffsetLocalChunkOutOfPlace(transaction::Transaction* transaction, common::node_group_idx_t nodeGroupIdx,
+        bool isNewNodeGroup, const ChunkCollection& localInsertChunks,
         const offset_to_row_idx_t& insertInfo, const ChunkCollection& localUpdateChunks,
-        const offset_to_row_idx_t& updateInfo);
+        const offset_to_row_idx_t& updateInfo, const offset_set_t& deleteInfo);
 
-    std::pair<std::unique_ptr<ChunkCollection>,std::unique_ptr<ChunkCollection>> extractSizeChunks(const ChunkCollection& localInsertChunks,
+    VarListInfo getVarListInfoForPrepareCommit(transaction::Transaction* transaction,common::node_group_idx_t nodeGroupIdx,  const ChunkCollection& localInsertChunks,
         const offset_to_row_idx_t& insertInfo, const ChunkCollection& localUpdateChunks,
         const offset_to_row_idx_t& updateInfo);
 
-    std::pair<std::unique_ptr<ChunkCollection>,std::unique_ptr<ChunkCollection>> extractOffsetChunks(transaction::Transaction* transaction,common::node_group_idx_t nodeGroupIdx,  const ChunkCollection& localInsertChunks,
-        const offset_to_row_idx_t& insertInfo, const ChunkCollection& localUpdateChunks,
-        const offset_to_row_idx_t& updateInfo);
+    void dataLocalChunkInPlaceCommit(common::node_group_idx_t nodeGroupIdx,
+        const ChunkCollection& localChunks);
+    void dataLocalChunkOutPlaceCommit(transaction::Transaction* transaction,
+        common::node_group_idx_t nodeGroupIdx, const ChunkCollection& localChunks);
+    bool canDataInPlaceCommit(transaction::Transaction* transaction,
+        common::node_group_idx_t nodeGroupIdx, const ChunkCollection& localChunks,
+        const common::offset_t totalListNum);
 
     void checkpointInMemory() final;
     void rollbackInMemory() final;

diff --git a/src/include/storage/store/var_list_column_chunk.h b/src/include/storage/store/var_list_column_chunk.h
@@ -95,6 +95,7 @@ class VarListColumnChunk final : public ColumnChunk {
     bool isOffsetSortedAscending(uint64_t startPos, uint64_t endPos) const;
 
     bool asendingRatio() const;
+    void showInfo();
 
 protected:
     void copyListValues(const common::list_entry_t& entry, common::ValueVector* dataVector);

diff --git a/src/storage/store/column.cpp b/src/storage/store/column.cpp
@@ -807,6 +807,7 @@ void Column::applyLocalChunkToColumnChunk(const ChunkCollection& localChunks,
     for (auto& [offsetInDstChunk, rowIdx] : updateInfo) {
         auto [chunkIdx, offsetInLocalChunk] =
             LocalChunkedGroupCollection::getChunkIdxAndOffsetInChunk(rowIdx);
+        KU_ASSERT(localChunks[chunkIdx]->getDataType().getLogicalTypeID()==columnChunk->getDataType().getLogicalTypeID());
         columnChunk->write(
             localChunks[chunkIdx], offsetInLocalChunk, offsetInDstChunk, 1 /* numValues */);
     }