Method checkpointInMemAndOnDisk

src/storage/table/csr_node_group.cpp:516–579

}

void CSRNodeGroup::checkpointInMemAndOnDisk(const UniqLock& lock, NodeGroupCheckpointState& state) {
    // TODO(Guodong): Should skip early here if no changes in the node group, so we avoid scanning
    // the csr header. Case: No insertions/deletions in persistent chunk and no in-mem chunks.
    auto& csrState = state.cast<CSRNodeGroupCheckpointState>();
    // Scan old csr header from disk and construct new csr header.
    persistentChunkGroup->cast<ChunkedCSRNodeGroup>().scanCSRHeader(*state.mm, csrState);
    csrState.newHeader =
        std::make_unique<InMemChunkedCSRHeader>(*state.mm, false, StorageConfig::NODE_GROUP_SIZE);
    // TODO(Guodong): Find max node offset in the node group.
    csrState.newHeader->setNumValues(StorageConfig::NODE_GROUP_SIZE);
    csrState.newHeader->copyFrom(*csrState.oldHeader);
    auto leafRegions = collectLeafRegionsAndCSRLength(lock, csrState);
    DASSERT(std::is_sorted(leafRegions.begin(), leafRegions.end(),
        [](const auto& a, const auto& b) { return a.regionIdx < b.regionIdx; }));
    const auto regionsToCheckpoint = mergeRegionsToCheckpoint(csrState, leafRegions);
    if (regionsToCheckpoint.empty()) {
        // No csr regions need to be checkpointed, meaning nothing is updated or deleted.
        // We should reset the version and update info of the persistent chunked group.
        persistentChunkGroup->resetVersionAndUpdateInfo();
        if (csrState.columnIDs.size() != persistentChunkGroup->getNumColumns()) {
            // The column set of the node group has changed. We need to re-create the persistent
            // chunked group.
            persistentChunkGroup = createNewPersistentChunkGroup(
                persistentChunkGroup->cast<ChunkedCSRNodeGroup>(), csrState);
        }
        return;
    }
    if (regionsToCheckpoint.size() == 1 &&
        regionsToCheckpoint[0].level > DEFAULT_PACKED_CSR_INFO.calibratorTreeHeight) {
        // Need to re-distribute all CSR regions in the node group.
        redistributeCSRRegions(csrState, leafRegions);
    } else {
        for (auto& region : regionsToCheckpoint) {
            csrState.newHeader->populateRegionCSROffsets(region, *csrState.oldHeader);
            // The left node offset of a region should always maintain stable across length and
            // offset changes.
            DASSERT(csrState.oldHeader->getStartCSROffset(region.leftNodeOffset) ==
                    csrState.newHeader->getStartCSROffset(region.leftNodeOffset));
        }
    }

    // Count tuples across ALL node offsets, not just the regions being checkpointed.
    // A region may have no changes (so it is not in regionsToCheckpoint) yet still hold tuples.
    // If we only count tuples in regionsToCheckpoint we may incorrectly conclude that the node
    // group is empty and set persistentChunkGroup to nullptr, losing those untouched tuples.
    uint64_t numTuplesAfterCheckpoint = 0;
    const auto numNodeOffsets = csrState.newHeader->length->getNumValues();
    for (auto i = 0u; i < numNodeOffsets; ++i) {
        numTuplesAfterCheckpoint += csrState.newHeader->getCSRLength(i);
    }
    if (numTuplesAfterCheckpoint == 0) {
        reclaimStorage(csrState.pageAllocator, lock);
        persistentChunkGroup = nullptr;
    } else {
        DASSERT(csrState.newHeader->sanityCheck());
        for (const auto columnID : csrState.columnIDs) {
            checkpointColumn(lock, columnID, csrState, regionsToCheckpoint);
        }