Method checkpointColumnInRegion

src/storage/table/csr_node_group.cpp:829–900

}

std::vector<ChunkCheckpointState> CSRNodeGroup::checkpointColumnInRegion(const UniqLock& lock,
    column_id_t columnID, const CSRNodeGroupCheckpointState& csrState,
    const CSRRegion& region) const {
    const auto* txn = csrState.transaction ? csrState.transaction : &DUMMY_CHECKPOINT_TRANSACTION;
    const auto leftCSROffset = csrState.oldHeader->getStartCSROffset(region.leftNodeOffset);
    DASSERT(leftCSROffset == csrState.newHeader->getStartCSROffset(region.leftNodeOffset));
    const auto rightCSROffset = csrState.oldHeader->getEndCSROffset(region.rightNodeOffset);
    const auto numOldRowsInRegion = rightCSROffset - leftCSROffset;

    Column* column = csrState.columns[columnID];
    LazySegmentScanner oldChunkScanner{*csrState.mm, column->getDataType().copy(),
        enableCompression};
    auto chunkState = scanCommittedUpdates(txn, persistentChunkGroup->getColumnChunk(columnID),
        column, oldChunkScanner, leftCSROffset, numOldRowsInRegion);

    const auto dummyChunkForNulls = ColumnChunkFactory::createColumnChunkData(*csrState.mm,
        dataTypes[columnID].copy(), false, DEFAULT_VECTOR_CAPACITY, ResidencyState::IN_MEMORY);
    dummyChunkForNulls->resetToAllNull();

    std::vector<ChunkCheckpointState> ret;

    CheckpointReadCursor readCursor{oldChunkScanner, leftCSROffset};
    CheckpointWriteCursor writeCursor{leftCSROffset, *csrState.mm, column->getDataType(), ret};

    // Copy per csr list from old chunk and merge with new insertions into the newChunkData.
    for (auto nodeOffset = region.leftNodeOffset; nodeOffset <= region.rightNodeOffset;
         nodeOffset++) {
        const auto oldCSRLength = csrState.oldHeader->getCSRLength(nodeOffset);

        DASSERT(csrState.newHeader->getStartCSROffset(nodeOffset) == writeCursor.getCSROffset());
        DASSERT(csrState.oldHeader->getStartCSROffset(nodeOffset) == readCursor.getCSROffset());

        // Copy old csr list with updates into the new chunk.
        if (!region.hasPersistentDeletions) {
            writeCSRListNoPersistentDeletions(readCursor, writeCursor, oldCSRLength);
        } else {
            writeCSRListWithPersistentDeletions(txn, readCursor, writeCursor, oldCSRLength,
                *persistentChunkGroup);
        }
        // Merge in-memory insertions into the new chunk.
        if (csrIndex) {
            auto rows = csrIndex->indices[nodeOffset].getRows();
            // TODO(Guodong): Optimize here. if no deletions and has sequential rows, scan in
            // range.
            for (const auto row : rows) {
                if (row == INVALID_ROW_IDX) {
                    continue;
                }
                auto [chunkIdx, rowInChunk] = StorageUtils::getQuotientRemainder(row,
                    StorageConfig::CHUNKED_NODE_GROUP_CAPACITY);
                const auto chunkedGroup = chunkedGroups.getGroup(lock, chunkIdx);
                writeInMemoryCSRInsertion(txn, writeCursor, *chunkedGroup, rowInChunk, columnID,
                    chunkState);
            }
        }

        const length_t numGaps = csrState.newHeader->getGapSize(nodeOffset);
        const length_t numOldGaps = csrState.oldHeader->getGapSize(nodeOffset);