From 052773e4db1045c130285ea3467775927bbc4e06 Mon Sep 17 00:00:00 2001 From: Darek Date: Sat, 17 Oct 2026 23:15:45 +0200 Subject: [PATCH] perf(storage): decode sparse CSR headers in clusters and bound the neighbour refill A rel scan decoded the CSR header span from the entry before a batch's first bound node to its last one. For bound nodes spread thinly over a node group that span is most of the header: LDBC SF1 q22 decoded 7.2M entries for 909 bound nodes. Every refill of the neighbour cache also copied the next 2048 CSR rows, whether the batch needed them or not. A batch whose span is at least 4x its bound nodes now decodes only clusters of the entries those nodes need, merging runs under 256 entries apart, in place in the full-size header buffer: at most 256 clusters, counted as one of the group's 16 partial decodes. The header records the decoded ranges and a whole-header flag, so a later batch re-decodes only when one of its nodes falls outside them, and nodes past the on-disk header read as empty lists. A batch out of offset order gets no clusters. The cluster list reuses the header's vector, and data-column scan states are initialized once per node group. A refill checks at most the next 4 bound lists, stops before the first that starts 256 or more rows past the lists before it, and otherwise fills the 2048-row window up to the batch's last list. It runs out of line from the loop that serves cached rows. With runtime checks the header buffer is poisoned on group entry and every header read asserts coverage. Co-Authored-By: Claude Opus 5.5 --- .../storage/table/csr_chunked_node_group.h | 10 + src/include/storage/table/csr_node_group.h | 9 +- src/storage/table/column_chunk.cpp | 4 +- src/storage/table/csr_chunked_node_group.cpp | 27 +- src/storage/table/csr_node_group.cpp | 364 +++++++++++++----- test/test_files/scan/csr_header_clusters.test | 120 ++++++ 6 files changed, 424 insertions(+), 110 deletions(-) create mode 100644 test/test_files/scan/csr_header_clusters.test diff --git a/src/include/storage/table/csr_chunked_node_group.h b/src/include/storage/table/csr_chunked_node_group.h index c4fb724fc..13df910a8 100644 --- a/src/include/storage/table/csr_chunked_node_group.h +++ b/src/include/storage/table/csr_chunked_node_group.h @@ -1,6 +1,8 @@ #pragma once #include +#include +#include #include "storage/enums/residency_state.h" #include "storage/table/chunked_node_group.h" @@ -62,6 +64,14 @@ struct LBUG_API InMemChunkedCSRHeader { bool randomLookup = false; // Logical header index stored at position 0. A full-group header uses 0. common::offset_t indexBase = 0; + // Scan-private cache: a full on-disk header, or the [start, end) index ranges last decoded + // for a bound-node batch. Nodes at or past headerLimit, the on-disk header size, have no + // lists and read as empty. Checkpoint headers leave these at the defaults. + bool wholeHeader = false; + common::offset_t headerLimit = 0; + std::vector> decodedRanges; + + bool coversNode(common::offset_t nodeOffset) const; InMemChunkedCSRHeader(MemoryManager& memoryManager, bool enableCompression, uint64_t capacity); InMemChunkedCSRHeader(std::unique_ptr offset, diff --git a/src/include/storage/table/csr_node_group.h b/src/include/storage/table/csr_node_group.h index c0f60e9e2..9e86e997f 100644 --- a/src/include/storage/table/csr_node_group.h +++ b/src/include/storage/table/csr_node_group.h @@ -145,7 +145,8 @@ struct CSRNodeGroupScanState final : NodeGroupScanState { // The number of rows (i.e., rels) that have been scanned so far in current node group. common::row_idx_t numCachedRows; common::row_idx_t nextCachedRowToScan; - // Entries and spans decoded so far for the current node group's header. + // Entries decoded and partial decodes (spans or cluster sets) so far for the current node + // group's header. common::offset_t numHeaderEntriesDecoded = 0; common::idx_t numHeaderSpansDecoded = 0; @@ -178,6 +179,12 @@ struct CSRNodeGroupScanState final : NodeGroupScanState { // node vector's chunk state, which holds the served parents in its selection vector. See // docs/multi_parent_lifetime.md. bool tryScanCachedTuplesPacked(RelTableScanState& tableScanState); + + // Copies rows into the cache starting in the current bound node's list and running through + // the next lists up to a gap (see refillEnd), or moves past that node when its list is empty + // or already scanned. False once the batch has no rows left. + bool refillCache(const transaction::Transaction* transaction, + RelTableScanState& tableScanState); }; struct CSRNodeGroupCheckpointState final : NodeGroupCheckpointState { diff --git a/src/storage/table/column_chunk.cpp b/src/storage/table/column_chunk.cpp index 5515f4b89..ca74eca55 100644 --- a/src/storage/table/column_chunk.cpp +++ b/src/storage/table/column_chunk.cpp @@ -90,7 +90,7 @@ void ColumnChunk::scan(const Transaction* transaction, const ChunkState& state, static void scanPersistentSegments(ChunkState& chunkState, ColumnChunkScanner& output, common::offset_t startRow, common::offset_t numRows) { - DASSERT(output.getNumValues() == 0); + [[maybe_unused]] const auto numValuesBefore = output.getNumValues(); [[maybe_unused]] uint64_t numValuesScanned = chunkState.rangeSegments(startRow, numRows, [&](auto& segmentState, auto offsetInSegment, auto lengthInSegment, auto) { output.scanSegment(offsetInSegment, lengthInSegment, @@ -100,7 +100,7 @@ static void scanPersistentSegments(ChunkState& chunkState, ColumnChunkScanner& o lengthInSegment); }); }); - DASSERT(output.getNumValues() == numValuesScanned); + DASSERT(output.getNumValues() == numValuesBefore + numValuesScanned); } void ColumnChunk::scanInMemSegments(ColumnChunkScanner& output, common::offset_t startRow, diff --git a/src/storage/table/csr_chunked_node_group.cpp b/src/storage/table/csr_chunked_node_group.cpp index 5c25224ca..cfa3d6560 100644 --- a/src/storage/table/csr_chunked_node_group.cpp +++ b/src/storage/table/csr_chunked_node_group.cpp @@ -398,6 +398,26 @@ InMemChunkedCSRHeader::InMemChunkedCSRHeader(MemoryManager& memoryManager, bool false /*initializeToZero*/); } +bool InMemChunkedCSRHeader::coversNode(offset_t nodeOffset) const { + if (randomLookup || wholeHeader || decodedRanges.empty()) { + return true; + } + if (headerLimit > 0 && nodeOffset >= headerLimit) { + return true; + } + const auto first = nodeOffset == 0 ? offset_t{0} : nodeOffset - 1; + const auto lastEx = nodeOffset + 1; + for (const auto& [start, end] : decodedRanges) { + if (start <= first && end >= lastEx) { + return true; + } + if (start > first) { + break; + } + } + return false; +} + offset_t InMemChunkedCSRHeader::getStartCSROffset(offset_t nodeOffset) const { // TODO(Guodong): I think we can simplify the check here by getting rid of some of the // conditions. @@ -408,8 +428,7 @@ offset_t InMemChunkedCSRHeader::getStartCSROffset(offset_t nodeOffset) const { if (randomLookup) { return offset->getValue(0); } - // A header decoded for a span is only read inside it. - DASSERT(indexBase == 0 || (nodeOffset > indexBase && nodeOffset - 1 - indexBase < numValues)); + DASSERT(coversNode(nodeOffset)); const auto stored = nodeOffset - 1 - indexBase; return offset->getValue(stored >= numValues ? (numValues - 1) : stored); } @@ -424,7 +443,7 @@ offset_t InMemChunkedCSRHeader::getEndCSROffset(offset_t nodeOffset) const { if (randomLookup) { return offset->getValue(nodeOffset == 0 ? 0 : 1); } - DASSERT(indexBase == 0 || nodeOffset - indexBase < numValues); + DASSERT(coversNode(nodeOffset)); const auto stored = nodeOffset - indexBase; return offset->getValue(stored >= numValues ? (numValues - 1) : stored); } @@ -433,7 +452,7 @@ length_t InMemChunkedCSRHeader::getCSRLength(offset_t nodeOffset) const { if (randomLookup) { return length->getNumValues() == 0 ? 0 : length->getValue(0); } - DASSERT(indexBase == 0 || nodeOffset - indexBase < length->getNumValues()); + DASSERT(coversNode(nodeOffset)); const auto stored = nodeOffset - indexBase; return stored >= length->getNumValues() ? 0 : length->getValue(stored); } diff --git a/src/storage/table/csr_node_group.cpp b/src/storage/table/csr_node_group.cpp index 0926369e9..c708574e2 100644 --- a/src/storage/table/csr_node_group.cpp +++ b/src/storage/table/csr_node_group.cpp @@ -1,5 +1,9 @@ #include "storage/table/csr_node_group.h" +#include +#include +#include + #include "common/constants.h" #include "common/serializer/buffer_reader.h" #include "common/serializer/buffer_writer.h" @@ -21,14 +25,21 @@ using namespace lbug::transaction; namespace lbug { namespace storage { -// Span decodes per node group before a scan decodes the group's whole CSR header. Each decode also -// re-initializes the column scan states, so batches that keep missing the span are cheaper with -// the whole header. +// Partial header decodes per node group before a scan decodes the group's whole CSR header. static constexpr idx_t MAX_HEADER_SPANS_PER_GROUP = 16; // After the first decode in a group, a batch with more bound nodes that are not spread thinly over // their span decodes the whole header: such batches come from scans that walk the whole group, // where one decode costs less than a span per batch. static constexpr idx_t MAX_SPAN_BATCH_SIZE = DEFAULT_VECTOR_CAPACITY / 16; +// Bound nodes whose header entries are less than CLUSTER_GAP apart share one decode: starting a +// cluster costs more than decoding the entries in between. +static constexpr offset_t CLUSTER_GAP = 256; +static constexpr idx_t MAX_CLUSTERS_PER_BATCH = 256; +// A neighbour-cache refill stops before a bound node whose list starts REFILL_GAP or more rows +// past the lists before it: copying those rows costs more than another refill. Only the next +// REFILL_LOOKAHEAD bound nodes are checked, so dense batches don't pay a header read per node. +static constexpr row_idx_t REFILL_GAP = 256; +static constexpr idx_t REFILL_LOOKAHEAD = 4; bool CSRNodeGroupScanState::tryScanCachedTuples(RelTableScanState& tableScanState) { if (numCachedRows == 0 || @@ -239,19 +250,6 @@ size_t CSRNodeGroup::getNumRetainedSnapshots() const { return retainedSnapshots.size(); } -static bool isWholeGroupHeader(const InMemChunkedCSRHeader& header, offset_t numGroupEntries) { - return header.indexBase == 0 && header.offset->getNumValues() == numGroupEntries; -} - -// A node's list starts at the offset stored for the node before it, so a span of the header -// covers the bound nodes when it holds that entry and the entry of the last bound node. -static bool spanCoversBoundNodes(const InMemChunkedCSRHeader& header, offset_t minNode, - offset_t maxNode) { - const auto firstNeeded = minNode == 0 ? offset_t{0} : minNode - 1; - return header.indexBase <= firstNeeded && - maxNode < header.indexBase + header.offset->getNumValues(); -} - static std::pair boundNodeOffsetRange(const RelTableScanState& relScanState) { offset_t minNode = StorageConfig::NODE_GROUP_SIZE; offset_t maxNode = 0; @@ -269,6 +267,97 @@ static std::pair boundNodeOffsetRange(const RelTableScanStat return {minNode, maxNode}; } +// One [start, end) of header indices per run of bound nodes whose needed entries sit within +// CLUSTER_GAP of each other. RelTable::initScanState sorts bound nodes by offset; a batch out of +// order or past MAX_CLUSTERS_PER_BATCH clusters gets none and decodes a span instead. Fills the +// caller's vector so its capacity carries over between decodes. +static void clustersForBoundNodes(const RelTableScanState& relScanState, offset_t headerSize, + std::vector>& clusters) { + clusters.clear(); + const auto& boundSel = relScanState.cachedBoundNodeSelVector; + for (sel_t i = 0; i < boundSel.getSelSize(); i++) { + const auto n = + relScanState.nodeIDVector->readNodeOffset(boundSel[i]) % StorageConfig::NODE_GROUP_SIZE; + if (n >= headerSize) { + continue; + } + const auto start = n == 0 ? offset_t{0} : n - 1; + const auto end = n + 1; + if (!clusters.empty() && start < clusters.back().first) { + clusters.clear(); + return; + } + if (!clusters.empty() && start < clusters.back().second + CLUSTER_GAP) { + clusters.back().second = std::max(clusters.back().second, end); + } else if (clusters.size() < MAX_CLUSTERS_PER_BATCH) { + clusters.emplace_back(start, end); + } else { + clusters.clear(); + return; + } + } +} + +static offset_t clusterEntryCount(const std::vector>& clusters) { + offset_t entries = 0; + for (const auto& [start, end] : clusters) { + entries += end - start; + } + return entries; +} + +static bool headerCoversBatch(const InMemChunkedCSRHeader& header, + const RelTableScanState& relScanState, offset_t numGroupEntries) { + if (header.wholeHeader) { + return true; + } + const auto& ranges = header.decodedRanges; + if (ranges.empty()) { + return false; + } + idx_t r = 0; + const auto& boundSel = relScanState.cachedBoundNodeSelVector; + for (sel_t i = 0; i < boundSel.getSelSize(); i++) { + const auto n = + relScanState.nodeIDVector->readNodeOffset(boundSel[i]) % StorageConfig::NODE_GROUP_SIZE; + if (n >= numGroupEntries) { + continue; + } + const auto first = n == 0 ? offset_t{0} : n - 1; + const auto lastEx = n + 1; + while (r < ranges.size() && ranges[r].second <= first) { + r++; + } + if (r >= ranges.size() || ranges[r].first > first || ranges[r].second < lastEx) { + return false; + } + } + return true; +} + +static void initDataColumnScanStates(const RelTableScanState& relScanState, + CSRNodeGroupScanState& nodeGroupScanState, ChunkedNodeGroup& persistent) { + for (auto i = 0u; i < relScanState.columnIDs.size(); i++) { + if (relScanState.columnIDs[i] == INVALID_COLUMN_ID || + relScanState.columnIDs[i] == ROW_IDX_COLUMN_ID) { + continue; + } + auto& chunk = persistent.getColumnChunk(relScanState.columnIDs[i]); + chunk.initializeScanState(nodeGroupScanState.chunkStates[i], relScanState.columns[i]); + } +} + +#if defined(RUNTIME_CHECKS) || !defined(NDEBUG) +// Reads past numEntries clamp to the last entry, so only the first numEntries can be stale. +static void poisonHeaderCache(InMemChunkedCSRHeader& header, offset_t numEntries) { + const auto num = std::min(numEntries, header.offset->getCapacity()); + std::fill(header.offset->getData(), header.offset->getData() + num, + std::numeric_limits::max()); + std::fill(header.length->getData(), header.length->getData() + num, + std::numeric_limits::max()); +} +#endif + void CSRNodeGroup::initializeScanState(const Transaction* transaction, TableScanState& state) const { auto& relScanState = state.cast(); @@ -286,6 +375,16 @@ void CSRNodeGroup::initializeScanState(const Transaction* transaction, if (nodeGroupScanState.pin.persistent) { nodeGroupScanState.numHeaderEntriesDecoded = 0; nodeGroupScanState.numHeaderSpansDecoded = 0; + initDataColumnScanStates(relScanState, nodeGroupScanState, + *nodeGroupScanState.pin.persistent); +#if defined(RUNTIME_CHECKS) || !defined(NDEBUG) + if (nodeGroupScanState.header) { + poisonHeaderCache(*nodeGroupScanState.header, + nodeGroupScanState.pin.persistent->cast() + .getCSRHeader() + .offset->getNumValues()); + } +#endif initScanForCommittedPersistent(transaction, relScanState, nodeGroupScanState); } } else if (relScanState.randomLookup && nodeGroupScanState.pin.persistent) { @@ -294,18 +393,14 @@ void CSRNodeGroup::initializeScanState(const Transaction* transaction, initScanForCommittedPersistent(transaction, relScanState, nodeGroupScanState); } else if (nodeGroupScanState.pin.persistent && relScanState.cachedBoundNodeSelVector.getSelSize() > 0) { - // A later batch of bound nodes can fall outside the header span decoded for an earlier - // one; decode again for it. + // A later batch of bound nodes can fall outside the header ranges decoded for an + // earlier one; decode again for it. const auto numGroupEntries = nodeGroupScanState.pin.persistent->cast() .getCSRHeader() .offset->getNumValues(); const auto& header = *nodeGroupScanState.header; - if (!isWholeGroupHeader(header, numGroupEntries)) { - const auto [minNode, maxNode] = boundNodeOffsetRange(relScanState); - const bool covered = spanCoversBoundNodes(header, minNode, maxNode); - if (!covered) { - initScanForCommittedPersistent(transaction, relScanState, nodeGroupScanState); - } + if (!headerCoversBatch(header, relScanState, numGroupEntries)) { + initScanForCommittedPersistent(transaction, relScanState, nodeGroupScanState); } } // Switch to a new Vector of bound nodes (i.e., new csr lists) in the node group. @@ -332,72 +427,89 @@ void CSRNodeGroup::initScanForCommittedPersistent(const Transaction* transaction auto& persistent = *nodeGroupScanState.pin.persistent; auto& csrChunkGroup = persistent.cast(); const auto& csrHeader = csrChunkGroup.getCSRHeader(); - // We are switching to a new node group. - // Initialize the scan states of a new node group for the csr header. csrHeader.offset->initializeScanState(offsetState, relScanState.csrOffsetColumn); csrHeader.length->initializeScanState(lengthState, relScanState.csrLengthColumn); - nodeGroupScanState.header->offset->setNumValues(0); - nodeGroupScanState.header->length->setNumValues(0); - // Initialize the scan states of a new node group for data columns. - for (auto i = 0u; i < relScanState.columnIDs.size(); i++) { - if (relScanState.columnIDs[i] == INVALID_COLUMN_ID || - relScanState.columnIDs[i] == ROW_IDX_COLUMN_ID) { - continue; - } - auto& chunk = persistent.getColumnChunk(relScanState.columnIDs[i]); - chunk.initializeScanState(nodeGroupScanState.chunkStates[i], relScanState.columns[i]); - } + auto& outHeader = *nodeGroupScanState.header; + outHeader.offset->setNumValues(0); + outHeader.length->setNumValues(0); + outHeader.indexBase = 0; + outHeader.wholeHeader = false; + outHeader.decodedRanges.clear(); DASSERT(csrHeader.offset->getNumValues() == csrHeader.length->getNumValues()); - nodeGroupScanState.header->indexBase = 0; + const auto numGroupEntries = csrHeader.offset->getNumValues(); + outHeader.headerLimit = numGroupEntries; if (relScanState.randomLookup) { auto pos = relScanState.nodeIDVector->state->getSelVector()[0]; auto nodeOffset = relScanState.nodeIDVector->readNodeOffset(pos); auto offsetInGroup = nodeOffset % StorageConfig::NODE_GROUP_SIZE; auto offsetToScanFrom = offsetInGroup == 0 ? 0 : offsetInGroup - 1; csrHeader.offset->scanCommitted(transaction, offsetState, - *nodeGroupScanState.header->offset, offsetToScanFrom, 1); + *outHeader.offset, offsetToScanFrom, 1); csrHeader.length->scanCommitted(transaction, lengthState, - *nodeGroupScanState.header->length, offsetInGroup, 1); - } else { - const auto numBoundNodes = csrHeader.offset->getNumValues(); - const auto numBatchNodes = relScanState.cachedBoundNodeSelVector.getSelSize(); - // Decode only the span of the header the batch's bound nodes need, while the group has had - // fewer than MAX_HEADER_SPANS_PER_GROUP spans and their entries stay below the whole - // header. Bound nodes past the header have no lists in this group. - bool decodeSpan = false; - offset_t indexBase = 0; - offset_t maxNode = 0; - if (numBatchNodes > 0 && - nodeGroupScanState.numHeaderSpansDecoded < MAX_HEADER_SPANS_PER_GROUP) { - const auto [minNode, lastNode] = boundNodeOffsetRange(relScanState); - indexBase = minNode == 0 ? 0 : minNode - 1; - maxNode = lastNode; - const auto span = maxNode - indexBase + 1; - decodeSpan = (nodeGroupScanState.numHeaderSpansDecoded == 0 || - numBatchNodes <= MAX_SPAN_BATCH_SIZE || numBatchNodes * 4 <= span) && - maxNode < numBoundNodes && - nodeGroupScanState.numHeaderEntriesDecoded + span < numBoundNodes; - } - if (!decodeSpan) { - csrHeader.offset->scanCommitted(transaction, offsetState, - *nodeGroupScanState.header->offset); - csrHeader.length->scanCommitted(transaction, lengthState, - *nodeGroupScanState.header->length); - nodeGroupScanState.numTotalRows = - nodeGroupScanState.header->getStartCSROffset(numBoundNodes); - } else { - const auto span = maxNode - indexBase + 1; - nodeGroupScanState.numHeaderEntriesDecoded += span; - nodeGroupScanState.numHeaderSpansDecoded++; - nodeGroupScanState.header->indexBase = indexBase; + *outHeader.length, offsetInGroup, 1); + outHeader.randomLookup = true; + return; + } + outHeader.randomLookup = false; + const auto numBatchNodes = relScanState.cachedBoundNodeSelVector.getSelSize(); + const auto [minNode, lastNode] = numBatchNodes > 0 ? boundNodeOffsetRange(relScanState) : + std::pair{0, 0}; + const auto indexBase = minNode == 0 ? offset_t{0} : minNode - 1; + const auto span = (numBatchNodes == 0 || lastNode >= numGroupEntries) ? + numGroupEntries : + lastNode - indexBase + 1; + const bool sparse = numBatchNodes * 4 <= span; + const bool laterOk = nodeGroupScanState.numHeaderSpansDecoded == 0 || + numBatchNodes <= MAX_SPAN_BATCH_SIZE || sparse; + const bool canPartial = numBatchNodes > 0 && lastNode < numGroupEntries && + nodeGroupScanState.numHeaderSpansDecoded < MAX_HEADER_SPANS_PER_GROUP && + laterOk; + // Clusters beat the span only for bound nodes spread thinly over it. + auto& clusters = outHeader.decodedRanges; + if (canPartial && sparse) { + clustersForBoundNodes(relScanState, numGroupEntries, clusters); + } + const auto clusterEntries = clusterEntryCount(clusters); + const bool decodeClusters = + canPartial && !clusters.empty() && clusters.size() * CLUSTER_GAP < span && + nodeGroupScanState.numHeaderEntriesDecoded + clusterEntries < numGroupEntries; + if (!decodeClusters) { + clusters.clear(); + } + const bool decodeSpan = canPartial && !decodeClusters && + nodeGroupScanState.numHeaderEntriesDecoded + span < numGroupEntries; + if (decodeClusters) { + nodeGroupScanState.numHeaderEntriesDecoded += clusterEntries; + nodeGroupScanState.numHeaderSpansDecoded++; + for (const auto& [start, end] : clusters) { + outHeader.offset->setNumValues(start); + outHeader.length->setNumValues(start); csrHeader.offset->scanCommitted(transaction, offsetState, - *nodeGroupScanState.header->offset, indexBase, span); + *outHeader.offset, start, end - start); csrHeader.length->scanCommitted(transaction, lengthState, - *nodeGroupScanState.header->length, indexBase, span); - nodeGroupScanState.numTotalRows = nodeGroupScanState.header->getEndCSROffset(maxNode); - } + *outHeader.length, start, end - start); + } + outHeader.offset->setNumValues(outHeader.decodedRanges.back().second); + outHeader.length->setNumValues(outHeader.decodedRanges.back().second); + nodeGroupScanState.numTotalRows = outHeader.getEndCSROffset(lastNode); + } else if (decodeSpan) { + nodeGroupScanState.numHeaderEntriesDecoded += span; + nodeGroupScanState.numHeaderSpansDecoded++; + outHeader.indexBase = indexBase; + outHeader.decodedRanges.emplace_back(indexBase, indexBase + span); + csrHeader.offset->scanCommitted(transaction, offsetState, + *outHeader.offset, indexBase, span); + csrHeader.length->scanCommitted(transaction, lengthState, + *outHeader.length, indexBase, span); + nodeGroupScanState.numTotalRows = outHeader.getEndCSROffset(lastNode); + } else { + outHeader.wholeHeader = true; + csrHeader.offset->scanCommitted(transaction, offsetState, + *outHeader.offset); + csrHeader.length->scanCommitted(transaction, lengthState, + *outHeader.length); + nodeGroupScanState.numTotalRows = outHeader.getStartCSROffset(numGroupEntries); } - nodeGroupScanState.header->randomLookup = relScanState.randomLookup; } void CSRNodeGroup::initScanForCommittedInMem(RelTableScanState& relScanState, @@ -452,6 +564,37 @@ NodeGroupScanResult CSRNodeGroup::scanCommittedPersistent(const Transaction* tra return scanCommittedPersistentWithCache(transaction, tableState, nodeGroupScanState); } +// End of a refill that starts in the current bound node's list. Without a gap among the next +// REFILL_LOOKAHEAD lists it fills the window, but never past the batch's last list. +static row_idx_t refillEnd(const RelTableScanState& tableState, const InMemChunkedCSRHeader& header, + row_idx_t listEnd, row_idx_t windowEnd) { + if (listEnd >= windowEnd) { + return windowEnd; + } + const auto& boundSel = tableState.cachedBoundNodeSelVector; + const auto offsetInGroup = [&](idx_t i) { + return tableState.nodeIDVector->readNodeOffset(boundSel[i]) % + StorageConfig::NODE_GROUP_SIZE; + }; + const idx_t numBound = boundSel.getSelSize(); + auto end = listEnd; + auto i = tableState.currBoundNodeIdx + 1; + for (; i < numBound && i <= tableState.currBoundNodeIdx + REFILL_LOOKAHEAD; i++) { + const auto node = offsetInGroup(i); + const auto start = header.getStartCSROffset(node); + if (start >= std::min(windowEnd, end + REFILL_GAP)) { + return std::min(end, windowEnd); + } + end = std::max(end, start + header.getCSRLength(node)); + } + if (i >= numBound) { + return std::min(end, windowEnd); + } + const auto lastNode = offsetInGroup(numBound - 1); + const auto lastEnd = header.getStartCSROffset(lastNode) + header.getCSRLength(lastNode); + return std::min(windowEnd, std::max(end, lastEnd)); +} + NodeGroupScanResult CSRNodeGroup::scanCommittedPersistentWithCache(const Transaction* transaction, RelTableScanState& tableState, CSRNodeGroupScanState& nodeGroupScanState) const { while (true) { @@ -464,36 +607,51 @@ NodeGroupScanResult CSRNodeGroup::scanCommittedPersistentWithCache(const Transac tableState.outState->getSelVector().getSelSize()}; } } - if (nodeGroupScanState.nextRowToScan == nodeGroupScanState.numTotalRows || - tableState.currBoundNodeIdx >= tableState.cachedBoundNodeSelVector.getSelSize()) { + if (!nodeGroupScanState.refillCache(transaction, tableState)) { return NODE_GROUP_SCAN_EMPTY_RESULT; } - const auto currNodeOffset = tableState.nodeIDVector->readNodeOffset( - tableState.cachedBoundNodeSelVector[tableState.currBoundNodeIdx]); - const auto offsetInGroup = currNodeOffset % StorageConfig::NODE_GROUP_SIZE; - const auto startCSROffset = nodeGroupScanState.header->getStartCSROffset(offsetInGroup); - if (startCSROffset > nodeGroupScanState.nextRowToScan) { - nodeGroupScanState.nextRowToScan = startCSROffset; - } - DASSERT(nodeGroupScanState.nextRowToScan <= nodeGroupScanState.numTotalRows); - const auto numToScan = - std::min(nodeGroupScanState.numTotalRows - nodeGroupScanState.nextRowToScan, - DEFAULT_VECTOR_CAPACITY); - nodeGroupScanState.pin.persistent->scan(transaction, tableState, nodeGroupScanState, - nodeGroupScanState.nextRowToScan, numToScan); - nodeGroupScanState.numCachedRows = numToScan; - nodeGroupScanState.nextRowToScan += numToScan; - if (tableState.outState->getSelVector().isUnfiltered()) { - nodeGroupScanState.cachedScannedVectorsSelBitset.reset(); - } else { - nodeGroupScanState.cachedScannedVectorsSelBitset = - std::bitset(); - for (auto i = 0u; i < tableState.outState->getSelVector().getSelSize(); i++) { - nodeGroupScanState.cachedScannedVectorsSelBitset->set( - tableState.outState->getSelVector()[i], true); - } + } +} + +bool CSRNodeGroupScanState::refillCache(const Transaction* transaction, + RelTableScanState& tableScanState) { + if (nextRowToScan == numTotalRows || + tableScanState.currBoundNodeIdx >= tableScanState.cachedBoundNodeSelVector.getSelSize()) { + return false; + } + const auto currNodeOffset = tableScanState.nodeIDVector->readNodeOffset( + tableScanState.cachedBoundNodeSelVector[tableScanState.currBoundNodeIdx]); + const auto offsetInGroup = currNodeOffset % StorageConfig::NODE_GROUP_SIZE; + const auto csrLength = header->getCSRLength(offsetInGroup); + if (csrLength == 0) { + tableScanState.currBoundNodeIdx++; + return true; + } + const auto startCSROffset = header->getStartCSROffset(offsetInGroup); + if (startCSROffset > nextRowToScan) { + nextRowToScan = startCSROffset; + } + DASSERT(nextRowToScan <= numTotalRows); + const auto listEnd = startCSROffset + csrLength; + if (listEnd <= nextRowToScan) { + tableScanState.currBoundNodeIdx++; + return true; + } + const auto windowEnd = + nextRowToScan + std::min(numTotalRows - nextRowToScan, DEFAULT_VECTOR_CAPACITY); + const auto numToScan = refillEnd(tableScanState, *header, listEnd, windowEnd) - nextRowToScan; + pin.persistent->scan(transaction, tableScanState, *this, nextRowToScan, numToScan); + numCachedRows = numToScan; + nextRowToScan += numToScan; + if (tableScanState.outState->getSelVector().isUnfiltered()) { + cachedScannedVectorsSelBitset.reset(); + } else { + cachedScannedVectorsSelBitset = std::bitset(); + for (auto i = 0u; i < tableScanState.outState->getSelVector().getSelSize(); i++) { + cachedScannedVectorsSelBitset->set(tableScanState.outState->getSelVector()[i], true); } } + return true; } NodeGroupScanResult CSRNodeGroup::scanCommittedPersistentWithoutCache( diff --git a/test/test_files/scan/csr_header_clusters.test b/test/test_files/scan/csr_header_clusters.test new file mode 100644 index 000000000..dfbee0b05 --- /dev/null +++ b/test/test_files/scan/csr_header_clusters.test @@ -0,0 +1,120 @@ +-DATASET CSV empty +-- + +-CASE CSRHeaderClusters +# One thread keeps each hub's F-neighbours as one bound-node batch for the E scan. +-STATEMENT CALL threads=1; +---- ok +-STATEMENT CREATE NODE TABLE N(id INT64, PRIMARY KEY(id)); +---- ok +-STATEMENT CREATE REL TABLE E(FROM N TO N); +---- ok +-STATEMENT CREATE REL TABLE F(FROM N TO N); +---- ok +-STATEMENT UNWIND range(0, 7999) AS i CREATE (:N {id: i}); +---- ok +-STATEMENT MATCH (a:N), (b:N) WHERE b.id = (a.id + 1) % 8000 CREATE (a)-[:E]->(b); +---- ok +# Hub 0: three clusters: node 0 with node 100, a middle node, and the last node. +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 0 AND a.id IN [0, 100, 4000, 7999] CREATE (h)-[:F]->(a); +---- ok +# Hub 1: one tight cluster, decoded as a packed span. +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 1 AND a.id IN [50, 51, 52] CREATE (h)-[:F]->(a); +---- ok +# Hub 2: two clusters with a large gap (nodes in the gap must not be read). +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 2 AND a.id IN [10, 5000] CREATE (h)-[:F]->(a); +---- ok +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 3 AND a.id = 0 CREATE (h)-[:F]->(a); +---- ok +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 4 AND a.id = 7999 CREATE (h)-[:F]->(a); +---- ok +# Hub 5: a dense run, decoded as a packed span. +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 5 AND a.id >= 1000 AND a.id < 1100 CREATE (h)-[:F]->(a); +---- ok +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 6 AND a.id IN [10, 11, 5000, 5001] CREATE (h)-[:F]->(a); +---- ok +-STATEMENT CHECKPOINT; +---- ok +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 0 RETURN count(*), sum(b.id); +---- 1 +4|4103 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 1 RETURN count(*), sum(b.id); +---- 1 +3|156 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 2 RETURN count(*), sum(b.id); +---- 1 +2|5012 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 3 RETURN count(*), sum(b.id); +---- 1 +1|1 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 4 RETURN count(*), sum(b.id); +---- 1 +1|0 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 5 RETURN count(*), sum(b.id); +---- 1 +100|105050 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 6 RETURN count(*), sum(b.id); +---- 1 +4|10026 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id <= 6 RETURN h.id, count(*), sum(b.id) ORDER BY h.id; +---- 7 +0|4|4103 +1|3|156 +2|2|5012 +3|1|1 +4|1|0 +5|100|105050 +6|4|10026 +# Insert inside a cluster after checkpoint. +-STATEMENT MATCH (a:N), (b:N) WHERE a.id = 50 AND b.id = 99 CREATE (a)-[:E]->(b); +---- ok +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 1 RETURN count(*), sum(b.id); +---- 1 +4|255 +# Delete a rel whose bound node sits in a cluster. +-STATEMENT MATCH (a:N)-[e:E]->(b:N) WHERE a.id = 10 AND b.id = 11 DELETE e; +---- ok +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 2 RETURN count(*), sum(b.id); +---- 1 +1|5001 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 6 RETURN count(*), sum(b.id); +---- 1 +3|10015 +# Bound nodes past the persistent header have no lists in this group. +-STATEMENT UNWIND range(8000, 8099) AS i CREATE (:N {id: i}); +---- ok +-STATEMENT MATCH (a:N) WHERE a.id >= 8000 MATCH (b:N {id: a.id - 8000}) CREATE (a)-[:E]->(b); +---- ok +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 2 AND a.id = 8000 CREATE (h)-[:F]->(a); +---- ok +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 2 RETURN count(*), sum(b.id); +---- 1 +2|5001 +# Hub 7's batch decodes two clusters; hub 8's batch, in the same scan, is covered by them except +# for node 8001 past the persistent header, which must read as an empty persistent list. +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 7 AND a.id IN [20, 5000] CREATE (h)-[:F]->(a); +---- ok +-STATEMENT MATCH (h:N), (a:N) WHERE h.id = 8 AND a.id IN [20, 8001] CREATE (h)-[:F]->(a); +---- ok +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id IN [7, 8] RETURN h.id, count(*), sum(b.id) ORDER BY h.id; +---- 2 +7|2|5022 +8|2|22 +-STATEMENT CHECKPOINT; +---- ok +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 0 RETURN count(*), sum(b.id); +---- 1 +4|4103 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 1 RETURN count(*), sum(b.id); +---- 1 +4|255 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 2 RETURN count(*), sum(b.id); +---- 1 +2|5001 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id = 6 RETURN count(*), sum(b.id); +---- 1 +3|10015 +-STATEMENT MATCH (h:N)-[:F]->(a:N)-[:E]->(b:N) WHERE h.id IN [7, 8] RETURN h.id, count(*), sum(b.id) ORDER BY h.id; +---- 2 +7|2|5022 +8|2|22