perf(storage): decode sparse CSR headers in clusters and bound the neighbour refill

src/include/storage/table/csr_chunked_node_group.h

@@ -1,6 +1,8 @@
 #pragma once
 
 #include <algorithm>
+#include <utility>
+#include <vector>
 
 #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<std::pair<common::offset_t, common::offset_t>> decodedRanges;
+
+    bool coversNode(common::offset_t nodeOffset) const;
 
     InMemChunkedCSRHeader(MemoryManager& memoryManager, bool enableCompression, uint64_t capacity);
     InMemChunkedCSRHeader(std::unique_ptr<ColumnChunkData> offset,

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 {

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,

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<offset_t>(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<offset_t>(stored >= numValues ? (numValues - 1) : stored);
 }
@@ -424,7 +443,7 @@ offset_t InMemChunkedCSRHeader::getEndCSROffset(offset_t nodeOffset) const {
     if (randomLookup) {
         return offset->getValue<offset_t>(nodeOffset == 0 ? 0 : 1);
     }
-    DASSERT(indexBase == 0 || nodeOffset - indexBase < numValues);
+    DASSERT(coversNode(nodeOffset));
     const auto stored = nodeOffset - indexBase;
     return offset->getValue<offset_t>(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<length_t>(0);
     }
-    DASSERT(indexBase == 0 || nodeOffset - indexBase < length->getNumValues());
+    DASSERT(coversNode(nodeOffset));
     const auto stored = nodeOffset - indexBase;
     return stored >= length->getNumValues() ? 0 : length->getValue<length_t>(stored);
 }