Implement multi-parent packed scan for CSR rel tables

.github/workflows/nightly-sanitizers.yml

@@ -16,6 +16,9 @@ jobs:
     timeout-minutes: 300
     steps:
       - uses: actions/checkout@v4
+      - name: Checkout dataset and benchmarks
+        run: |
+          git submodule update --init dataset benchmark # needed for demo db
       - name: Install build dependencies
         run: |
           sudo apt-get update
@@ -35,6 +38,9 @@ jobs:
     continue-on-error: true
     steps:
       - uses: actions/checkout@v4
+      - name: Checkout dataset and benchmarks
+        run: |
+          git submodule update --init dataset benchmark # needed for demo db
       - name: Install build dependencies
         run: |
           sudo apt-get update

docs/multi_parent_lifetime.md

@@ -1,5 +1,17 @@
 # Multi-parent packed child slices: lifetime & representation
 
+> **Status update (multi-parent scan implemented):** the deferred change described below has
+> been implemented. `PackedChildSlices` now holds a `shared_ptr<SelectionVector>` aliasing the
+> bound (parent) chunk state's selection vector plus a prefix-sum `offsets` over ALL served
+> parents (zeros allowed, zero-length ranges skipped by consumers). The CSR scan packs children
+> of multiple parents per output batch in `CSRNodeGroupScanState::tryScanCachedTuplesPacked`
+> (persistent with-cache path only). Multi-parent batches are enabled per-consumer:
+> `ScanRelTable::setMultiParentPackedScanEnabled(true)` is set by the plan mapper only when the
+> packed-aware `PackedFilteredCount` consumes the scan output directly. Plans that materialize
+> the scan output across a `FactorizedTable` (e.g. when the filter predicate touches an nbr
+> node property, which routes the property fetch through a hash join) keep the
+> one-parent-per-batch contract — exactly the materialization hazard described below.
+
 Context: the `enable_packed_path_extend` path reuses the adjacency-list (CSR)
 rel scan to implement a physical packed extend. `ScanRelTable` attaches a
 `PackedChildSlices` descriptor to the **output** `DataChunkState`

src/include/common/data_chunk/data_chunk_state.h

@@ -16,41 +16,34 @@ enum class FStateType : uint8_t {
 
 class LBUG_API DataChunkState {
 public:
+    // Describes how the children in the current output batch map back to their parents. Used by
+    // the PACKED_EXTEND physical operator; see docs/multi_parent_lifetime.md for the full
+    // lifetime & representation rationale.
+    //
+    // Representation (multi-parent packed scan):
+    //   - parentSelVector aliases the bound (parent) chunk state's selection vector for the
+    //     current output batch. The shared_ptr keeps the SelectionVector OBJECT alive even if
+    //     the input state is reused, but the buffer contents are NOT snapshotted: they are
+    //     whatever the bound chunk's selection vector currently holds (the parents whose
+    //     children are materialized in this batch).
+    //   - offsets is a prefix sum over ALL parents in parentSelVector, with
+    //     offsets.size() == parentSelVector->getSelSize() + 1; the children of parent i occupy
+    //     output positions [offsets[i], offsets[i+1]) in the child chunk's selection vector.
+    //     offsets[i] == offsets[i+1] means parent i produced no children in this batch
+    //     (consumers must skip zero-length ranges).
+    //
+    // Lifetime rule: the descriptor is valid only for synchronous consumption of this output
+    // batch. Do not persist it across a materialization boundary (e.g. appending to a
+    // FactorizedTable and reading back later): the aliased buffer is rewritten in place by the
+    // next input batch (setToFiltered/setToUnfiltered) and the descriptor is cleared/reset
+    // (ResultSet::resetForReuse, next scan call).
     struct PackedChildSlices {
-        std::vector<sel_t> parentPositions;
+        std::shared_ptr<SelectionVector> parentSelVector;
         std::vector<sel_t> offsets;
 
-        void clear() {
-            parentPositions.clear();
-            offsets.clear();
-        }
-
-        bool empty() const { return parentPositions.empty(); }
-        sel_t getNumParents() const { return parentPositions.size(); }
+        bool empty() const { return parentSelVector == nullptr; }
+        sel_t getNumParents() const { return empty() ? 0 : parentSelVector->getSelSize(); }
         sel_t getNumValues() const { return offsets.empty() ? 0 : offsets.back(); }
-
-        // Pre-allocate for an expected number of parents. Call this before a sequence of
-        // append() calls so each append is O(1) amortized with no reallocation.
-        // offsets holds one more entry than parentPositions (prefix-sum invariant), so reserve
-        // numParents+1 for it.
-        void reserve(size_t numParents) {
-            parentPositions.reserve(numParents);
-            offsets.reserve(numParents + 1);
-        }
-
-        // Append a parent slice: parent position and number of values for that parent.
-        // Maintains the invariant offsets.size() == parentPositions.size() + 1
-        void append(sel_t parentPosition, sel_t numValues) {
-            if (offsets.empty()) {
-                // initialize offsets with {0, numValues}
-                parentPositions.push_back(parentPosition);
-                offsets.push_back(0);
-                offsets.push_back(numValues);
-                return;
-            }
-            parentPositions.push_back(parentPosition);
-            offsets.push_back(offsets.back() + numValues);
-        }
     };
 
     DataChunkState();
@@ -79,30 +72,17 @@ class LBUG_API DataChunkState {
         DASSERT(packedChildSlices.has_value());
         return *packedChildSlices;
     }
-    void setPackedChildSlices(std::vector<sel_t> parentPositions, std::vector<sel_t> offsets) {
-        DASSERT(offsets.size() == parentPositions.size() + 1);
-        packedChildSlices = PackedChildSlices{std::move(parentPositions), std::move(offsets)};
-    }
-    void setSingleParentPackedChildSlice(sel_t parentPosition, sel_t numValues) {
-        setPackedChildSlices({parentPosition}, {0, numValues});
+    void setPackedChildSlices(std::shared_ptr<SelectionVector> parentSelVector,
+        std::vector<sel_t> offsets) {
+        DASSERT(parentSelVector != nullptr);
+        DASSERT(offsets.size() == parentSelVector->getSelSize() + 1);
+        packedChildSlices = PackedChildSlices{std::move(parentSelVector), std::move(offsets)};
     }
-
-    // Append a packed child slice for a parent. Creates packedChildSlices if not present.
-    void appendPackedChildSlice(sel_t parentPosition, sel_t numValues) {
-        if (!packedChildSlices.has_value()) {
-            setSingleParentPackedChildSlice(parentPosition, numValues);
-            return;
-        }
-        packedChildSlices->append(parentPosition, numValues);
-    }
-
-    // Pre-allocate the packed child slices for an expected number of parents. Creates the
-    // optional if not present so subsequent appendPackedChildSlice() calls don't reallocate.
-    void reservePackedChildSlices(size_t numParents) {
-        if (!packedChildSlices.has_value()) {
-            packedChildSlices = PackedChildSlices{};
-        }
-        packedChildSlices->reserve(numParents);
+    // Single-parent convenience: parentSelVector must hold exactly one parent position.
+    void setSingleParentPackedChildSlice(std::shared_ptr<SelectionVector> parentSelVector,
+        sel_t numValues) {
+        DASSERT(parentSelVector->getSelSize() == 1);
+        setPackedChildSlices(std::move(parentSelVector), {0, numValues});
     }
 
     void clearPackedChildSlices() { packedChildSlices.reset(); }

src/include/processor/operator/scan/scan_rel_table.h

@@ -99,28 +99,40 @@ class ScanRelTable final : public ScanTable {
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
+        std::unique_ptr<ScanRelTable> result;
         if (sourceMode) {
             if (sourceNodeScanMode) {
-                return std::make_unique<ScanRelTable>(opInfo.copy(), tableInfo.copy(),
+                result = std::make_unique<ScanRelTable>(opInfo.copy(), tableInfo.copy(),
                     copyVector(sourceNodeTableInfos), sourceNodeSharedStates,
                     sourceNodeProgressSharedState, sourceNodeScanInfo.copy(), id, printInfo->copy(),
                     operatorType);
+            } else {
+                result = std::make_unique<ScanRelTable>(opInfo.copy(), tableInfo.copy(),
+                    sourceNodeTables, id, printInfo->copy(), operatorType);
             }
-            return std::make_unique<ScanRelTable>(opInfo.copy(), tableInfo.copy(), sourceNodeTables,
-                id, printInfo->copy(), operatorType);
+        } else {
+            result = std::make_unique<ScanRelTable>(opInfo.copy(), tableInfo.copy(),
+                children[0]->copy(), id, printInfo->copy(), operatorType);
         }
-        return std::make_unique<ScanRelTable>(opInfo.copy(), tableInfo.copy(), children[0]->copy(),
-            id, printInfo->copy(), operatorType);
+        result->multiParentPackedScanEnabled = multiParentPackedScanEnabled;
+        return result;
     }
 
 protected:
     void initGlobalStateInternal(ExecutionContext* context) override;
     bool fetchNextBoundNodeBatch(transaction::Transaction* transaction);
     void updatePackedChildSlices(common::sel_t outputSize) const;
-    // Pre-allocate packedChildSlices for the current input batch. The number of parents that will
-    // be processed in this batch is known up front from cachedBoundNodeSelVector, so we reserve
-    // once per batch to keep subsequent appendPackedChildSlice() calls reallocation-free.
-    void reservePackedChildSlicesForBatch() const;
+
+    // Multi-parent packed batches (see docs/multi_parent_lifetime.md). Disabled by default:
+    // standard consumers of the packed extend output rely on the one-parent-per-batch
+    // factorization contract (bound vector flat with selSize 1). Only packed-aware consumers
+    // (currently PackedFilteredCount, which reads the PackedChildSlices descriptor) enable
+    // this, via setMultiParentPackedScanEnabled(), so the CSR scan may pack children of many
+    // parents into one output batch.
+    bool multiParentPackedScanEnabled = false;
+
+public:
+    void setMultiParentPackedScanEnabled(bool enabled) { multiParentPackedScanEnabled = enabled; }
 
 protected:
     ScanRelTableInfo tableInfo;

src/include/storage/table/csr_node_group.h

@@ -143,6 +143,14 @@ struct CSRNodeGroupScanState final : NodeGroupScanState {
     }
 
     bool tryScanCachedTuples(RelTableScanState& tableScanState);
+
+    // Multi-parent variant of tryScanCachedTuples (packedMultiParentScan): serves the children
+    // of as many consecutive parents as fit into one output batch. On success the batch is
+    // described by the output chunk's (filtered) selection vector plus
+    // RelTableScanState::packedChildOffsets (prefix sum over the served parents) and the bound
+    // node vector's chunk state, which holds the served parents in its selection vector. See
+    // docs/multi_parent_lifetime.md.
+    bool tryScanCachedTuplesPacked(RelTableScanState& tableScanState);
 };
 
 struct CSRNodeGroupCheckpointState final : NodeGroupCheckpointState {

src/include/storage/table/rel_table.h

@@ -28,6 +28,17 @@ struct RelTableScanState : TableScanState {
     // This is a reference of the original selVector of the input boundNodeIDVector.
     common::SelectionVector cachedBoundNodeSelVector;
 
+    // Multi-parent packed scan support (see docs/multi_parent_lifetime.md). When
+    // packedMultiParentScan is true, the CSR scan may serve children of MULTIPLE bound parents
+    // in a single output batch: the bound node vector's chunk state is set to unflat with its
+    // selection vector holding the served parents, and packedChildOffsets holds the prefix-sum
+    // offsets over those parents (offsets.size() == numServedParents + 1; zero-length ranges
+    // are skipped by consumers). Only the persistent with-cache CSR scan path packs multiple
+    // parents; all other paths keep the one-parent-per-batch contract (bound vector flat with
+    // selSize 1, packedChildOffsets left empty).
+    bool packedMultiParentScan = false;
+    std::vector<common::sel_t> packedChildOffsets;
+
     std::unique_ptr<LocalRelTableScanState> localTableScanState;
 
     // Optional state used by Arrow-backed relationship tables. Keep it on the common scan state so

src/processor/map/map_aggregate.cpp

@@ -17,6 +17,7 @@
 #include "processor/operator/aggregate/packed_filtered_count.h"
 #include "processor/operator/aggregate/simple_aggregate.h"
 #include "processor/operator/aggregate/simple_aggregate_scan.h"
+#include "processor/operator/scan/scan_rel_table.h"
 #include "processor/plan_mapper.h"
 #include "processor/result/result_set_descriptor.h"
 
@@ -231,8 +232,17 @@ static std::unique_ptr<PhysicalOperator> tryMapPackedFilteredCount(PlanMapper& m
         DataPos{packedChildSchema->getExpressionPos(*predicateInputs->first)},
         DataPos{packedChildSchema->getExpressionPos(*predicateInputs->second)},
         dependentGroupsVector[0], dependentGroupsVector[1], std::move(multiplicityChunks)};
+    auto packedChildPhysicalOp = mapper.mapOperator(packedChild);
+    // Enable multi-parent packed batches on the underlying rel scan (if it is a single-table
+    // ScanRelTable): PackedFilteredCount is packed-aware and consumes the PackedChildSlices
+    // descriptor, so it can attribute children of many parents per batch. Other consumers of
+    // packed extend output keep the one-parent-per-batch contract. See
+    // docs/multi_parent_lifetime.md.
+    if (auto* scanRelTable = dynamic_cast<ScanRelTable*>(packedChildPhysicalOp.get())) {
+        scanRelTable->setMultiParentPackedScanEnabled(true);
+    }
     auto sink = std::make_unique<PackedFilteredCount>(sharedState, info,
-        mapper.mapOperator(packedChild), mapper.getOperatorID(),
+        std::move(packedChildPhysicalOp), mapper.getOperatorID(),
         std::make_unique<PackedFilteredCountPrintInfo>(logicalFilter.getPredicate(),
             agg.getKeys()));
     sink->setDescriptor(std::make_unique<ResultSetDescriptor>(packedChildSchema));

src/processor/operator/aggregate/packed_filtered_count.cpp

@@ -64,13 +64,47 @@ uint64_t PackedFilteredCount::countMatchesForCurrentTuple() {
     for (auto* state : multiplicityStates) {
         baseMultiplicity *= state->getSelSize();
     }
-    if (baseMultiplicity == 0 || selectState->getSelSize() == 0 || flatState->getSelSize() == 0) {
-        return 0;
-    }
-
     uint64_t result = 0;
     const auto& lhsSelVector = lhsValueVector->state->getSelVector();
     const auto& rhsSelVector = rhsValueVector->state->getSelVector();
+    const auto packed = rhsValueVector->state->hasPackedChildSlices();
+    if (packed) {
+        // Multi-parent packed batch: the child chunk state carries a PackedChildSlices
+        // descriptor whose parentSelVector aliases the bound (parent) chunk's selection vector
+        // and whose offsets prefix-sum the children per parent (zero-length ranges are parents
+        // without children in this batch and are skipped). The lhs and group key live in the
+        // same (bound) chunk as the parent selection, and the child range
+        // [offsets[p], offsets[p+1]) indexes into the child chunk's selection vector. Per-parent
+        // counts are accumulated into localCounts here (the batch spans multiple group keys, so
+        // the caller cannot attribute the returned total to a single key). See
+        // docs/multi_parent_lifetime.md.
+        const auto& slices = rhsValueVector->state->getPackedChildSlices();
+        const auto& parentSelVector = *slices.parentSelVector;
+        for (sel_t parentIdx = 0; parentIdx < parentSelVector.getSelSize(); ++parentIdx) {
+            const auto start = slices.offsets[parentIdx];
+            const auto end = slices.offsets[parentIdx + 1];
+            if (start == end) {
+                continue;
+            }
+            const auto lhsValue = lhsValueVector->getValue<int64_t>(parentSelVector[parentIdx]);
+            uint64_t parentCount = 0;
+            for (auto rhsIdx = start; rhsIdx < end; ++rhsIdx) {
+                const auto rhsValue = rhsValueVector->getValue<int64_t>(rhsSelVector[rhsIdx]);
+                if ((lhsValue + rhsValue) % 10 == 0) {
+                    parentCount += baseMultiplicity;
+                }
+            }
+            if (parentCount > 0) {
+                localCounts[groupKeyVector->getValue<int64_t>(parentSelVector[parentIdx])] +=
+                    parentCount;
+                result += parentCount;
+            }
+        }
+        return result;
+    }
+    if (baseMultiplicity == 0 || selectState->getSelSize() == 0 || flatState->getSelSize() == 0) {
+        return 0;
+    }
     for (auto lhsIdx = 0u; lhsIdx < lhsSelVector.getSelSize(); ++lhsIdx) {
         const auto lhsValue = lhsValueVector->getValue<int64_t>(lhsSelVector[lhsIdx]);
         for (auto rhsIdx = 0u; rhsIdx < rhsSelVector.getSelSize(); ++rhsIdx) {
@@ -85,9 +119,13 @@ uint64_t PackedFilteredCount::countMatchesForCurrentTuple() {
 
 void PackedFilteredCount::executeInternal(ExecutionContext* context) {
     while (children[0]->getNextTuple(context)) {
-        const auto groupKeyPos = groupKeyVector->state->getSelVector()[0];
+        const auto packed = rhsValueVector->state->hasPackedChildSlices();
         const auto count = countMatchesForCurrentTuple();
-        if (count > 0) {
+        if (count > 0 && !packed) {
+            // Single-parent batch: attribute the whole batch's count to the one group key. For
+            // multi-parent packed batches, countMatchesForCurrentTuple() has already attributed
+            // per-parent counts to each parent's group key via localCounts.
+            const auto groupKeyPos = groupKeyVector->state->getSelVector()[0];
             localCounts[groupKeyVector->getValue<int64_t>(groupKeyPos)] += count;
         }
         metrics->numOutputTuple.incrementByOne();