From df60eb3797675fafeb7d85cb9e2209bd4f8d001d Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sun, 30 Aug 2026 09:27:51 -0700 Subject: [PATCH 1/4] Checkout dataset and benchmark submodules in nightly sanitizer workflow The sanitizer test suite needs the dataset submodule for the demo db; mirror the pattern used in ci-workflow.yml. --- .github/workflows/nightly-sanitizers.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/nightly-sanitizers.yml b/.github/workflows/nightly-sanitizers.yml index 28ae1c5b1..8b095d762 100644 --- a/.github/workflows/nightly-sanitizers.yml +++ b/.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 From c71743c36d95b8ca24493455d0ca93923b7f367c Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sun, 30 Aug 2026 12:04:15 -0700 Subject: [PATCH 2/4] Document packed child slice lifetime & representation contract Implement docs/multi_parent_lifetime.md's "for now" recommendation: - Keep the owned std::vector parentPositions copy in PackedChildSlices; explicitly document why a pointer/alias into the parent's SelectionVector is unsafe (contents rewritten in place by setToFiltered/setToUnfiltered). - Document the synchronous-consumption lifetime rule: the descriptor is valid only for the current output batch and must not be persisted across a materialization boundary. - Record the deferred multi-parent convention (shared_ptr sel vector + prefix-sum offsets over all parents, zeros allowed) for when the multi-parent packed scan is implemented. - Add a unit test codifying the owned-copy invariant against in-place sel vector mutation. --- .../common/data_chunk/data_chunk_state.h | 24 +++++++++++++++ .../operator/scan/scan_rel_table.cpp | 3 ++ test/planner/cardinality_test.cpp | 30 +++++++++++++++++++ 3 files changed, 57 insertions(+) diff --git a/src/include/common/data_chunk/data_chunk_state.h b/src/include/common/data_chunk/data_chunk_state.h index 6ecbafc38..34f35d393 100644 --- a/src/include/common/data_chunk/data_chunk_state.h +++ b/src/include/common/data_chunk/data_chunk_state.h @@ -16,6 +16,30 @@ 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 (current CSR scan processes ONE parent per output batch): + // - parentPositions is an OWNED copy holding only the parents that produced children + // (parents with no matches are dropped at scan time). It must never be replaced by a + // pointer/alias into the parent's SelectionVector: the vector object is shared_ptr-held + // but its contents are rewritten in place (setToFiltered/setToUnfiltered), so an alias + // would silently track whatever the input batch holds next. + // - offsets is a prefix sum with offsets.size() == parentPositions.size() + 1; the + // children of parent p occupy output positions [offsets[p], offsets[p+1]). + // + // 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 input batch advances and the descriptor is + // cleared/reset (ResultSet::resetForReuse, next scan call). + // + // Deferred (true multi-parent packed scan, many parents per output batch): point at the + // parent's selection vector kept alive via shared_ptr (getSelVectorShared()), include ALL + // parents (zero-child ones too), and switch offsets to a prefix sum over all parents + // (offsets[i] == offsets[i+1] means parent i has no children) so the consumer skips + // zero-length ranges. Until that scan exists, the owned-copy representation below is the + // contract. struct PackedChildSlices { std::vector parentPositions; std::vector offsets; diff --git a/src/processor/operator/scan/scan_rel_table.cpp b/src/processor/operator/scan/scan_rel_table.cpp index 52b003cf2..a77b6bff2 100644 --- a/src/processor/operator/scan/scan_rel_table.cpp +++ b/src/processor/operator/scan/scan_rel_table.cpp @@ -226,6 +226,9 @@ void ScanRelTable::updatePackedChildSlices(sel_t outputSize) const { scanState->outState->clearPackedChildSlices(); return; } + // See docs/multi_parent_lifetime.md for the representation/lifetime contract of the + // descriptor written here (owned copy; synchronous consumption only). + // // The CSR scan sets nodeIDVector to flat, pointing its selVector[0] at the actual parent // whose children are currently materialized in the output vector (see // RelTableScanState::setNodeIDVectorToFlat). We must use that position as the parent diff --git a/test/planner/cardinality_test.cpp b/test/planner/cardinality_test.cpp index 76b4c60fc..efce73eae 100644 --- a/test/planner/cardinality_test.cpp +++ b/test/planner/cardinality_test.cpp @@ -290,5 +290,35 @@ TEST_F(CardinalityTest, TestPackedChildSliceAppend) { } } +TEST_F(CardinalityTest, TestPackedChildSliceOwnedCopyLifetime) { + common::DataChunkState state; + // Simulate the CSR scan: the bound-node (parent) selection vector is flat with selSize 1, + // pointing its position 0 at parent row 9. + auto& selVector = state.getSelVectorUnsafe(); + selVector.setToFiltered(1); + selVector[0] = 9; + + state.setSingleParentPackedChildSlice(selVector[0], 5); + { + const auto& slices = state.getPackedChildSlices(); + ASSERT_EQ(1, slices.getNumParents()); + EXPECT_EQ(9, slices.parentPositions[0]); + EXPECT_EQ(5, slices.getNumValues()); + } + + // The selection vector's contents are mutable in place (setToFiltered/setToUnfiltered + // rewrite the buffer). Rewriting it here is what the next input batch would do. Since + // parentPositions is an OWNED copy — not an alias into the selection vector — the + // descriptor must keep the parent position recorded for this batch. See + // docs/multi_parent_lifetime.md for the synchronous-consumption lifetime rule. + selVector[0] = 42; + selVector.setToUnfiltered(1); + { + const auto& slices = state.getPackedChildSlices(); + EXPECT_EQ(9, slices.parentPositions[0]); + EXPECT_EQ(5, slices.getNumValues()); + } +} + } // namespace testing } // namespace lbug From eb567217d1309e77e73fee8c44d78a99373cecaf Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sun, 30 Aug 2026 13:48:24 -0700 Subject: [PATCH 3/4] Implement multi-parent packed scan for CSR rel tables Implement the deferred design from docs/multi_parent_lifetime.md: Representation (PackedChildSlices): - Replace the owned parentPositions copy with a shared_ptr alias to the bound (parent) chunk state's selection vector plus a prefix-sum offsets vector over ALL served parents (offsets.size() == parentSelSize + 1). Zero-length ranges (parents without children in the batch) are allowed; consumers skip them. The shared_ptr keeps the SelectionVector object alive while still aliasing mutable contents, preserving the synchronous-consumption lifetime rule. Storage (CSRNodeGroup): - Add tryScanCachedTuplesPacked: serves children of as many consecutive parents as fit into one output batch from the row cache, building the combined (filtered) output selection vector, the per-parent prefix sum in RelTableScanState::packedChildOffsets, and the served-parent selection on the bound chunk state (switched to unflat). Handles zero-length lists, capacity-limited partial lists, and cache-window boundaries; parents whose lists span batches are resumed seamlessly. - Single-parent serves (without-cache, in-memory, local storage) keep the existing one-parent-per-batch contract. Operator (ScanRelTable): - Attach the descriptor for both single- and multi-parent batches; the batch shape is derived from the bound vector's selection size. - Drop the speculative append/reserve machinery (superseded by direct offsets construction). Consumer gating: - Multi-parent batches change the factorization contract (bound chunk unflat with the served parents), so they are opt-in per consumer: the plan mapper enables it only for PackedFilteredCount consuming the scan output DIRECTLY (no FactorizedTable/hash-join between, the exact materialization hazard the doc warns about). Plans that fetch nbr node properties through the property-join pipeline keep single-parent batches. PackedFilteredCount: - Attribute counts per parent via the packed slices when present (a batch may span many group keys); keep the legacy single-tuple path when no descriptor is attached. Tests (test/planner/cardinality_test.cpp): - Rework the DataChunkState unit tests for the new representation and document the aliasing semantics in a dedicated test. - Add a storage-level test driving RelTable::scan directly with packing enabled over a 3000-node/6750-edge graph (degree 3, with zero-child parents): asserts multi-parent batches actually occur, the offsets prefix-sum invariants hold, and every edge is served exactly once with correct parent attribution. - Add SQL-level equivalence: the PackedFilteredCount pattern over an nbr property (join pipeline, single-parent batches) produces identical results with the packed extend on and off. Docs: mark the deferred design as implemented in docs/multi_parent_lifetime.md, noting the materialization boundary constraint that keeps the property-join plan shape single-parent. --- docs/multi_parent_lifetime.md | 12 + .../common/data_chunk/data_chunk_state.h | 98 ++---- .../processor/operator/scan/scan_rel_table.h | 30 +- src/include/storage/table/csr_node_group.h | 8 + src/include/storage/table/rel_table.h | 11 + src/processor/map/map_aggregate.cpp | 12 +- .../aggregate/packed_filtered_count.cpp | 50 ++- .../operator/scan/scan_rel_table.cpp | 57 ++-- src/storage/table/csr_node_group.cpp | 94 +++++- test/planner/cardinality_test.cpp | 316 ++++++++++++++---- 10 files changed, 490 insertions(+), 198 deletions(-) diff --git a/docs/multi_parent_lifetime.md b/docs/multi_parent_lifetime.md index aed9d66fe..46e55e163 100644 --- a/docs/multi_parent_lifetime.md +++ b/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` 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` diff --git a/src/include/common/data_chunk/data_chunk_state.h b/src/include/common/data_chunk/data_chunk_state.h index 34f35d393..43c683587 100644 --- a/src/include/common/data_chunk/data_chunk_state.h +++ b/src/include/common/data_chunk/data_chunk_state.h @@ -20,61 +20,30 @@ class LBUG_API DataChunkState { // the PACKED_EXTEND physical operator; see docs/multi_parent_lifetime.md for the full // lifetime & representation rationale. // - // Representation (current CSR scan processes ONE parent per output batch): - // - parentPositions is an OWNED copy holding only the parents that produced children - // (parents with no matches are dropped at scan time). It must never be replaced by a - // pointer/alias into the parent's SelectionVector: the vector object is shared_ptr-held - // but its contents are rewritten in place (setToFiltered/setToUnfiltered), so an alias - // would silently track whatever the input batch holds next. - // - offsets is a prefix sum with offsets.size() == parentPositions.size() + 1; the - // children of parent p occupy output positions [offsets[p], offsets[p+1]). + // 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 input batch advances and the descriptor is - // cleared/reset (ResultSet::resetForReuse, next scan call). - // - // Deferred (true multi-parent packed scan, many parents per output batch): point at the - // parent's selection vector kept alive via shared_ptr (getSelVectorShared()), include ALL - // parents (zero-child ones too), and switch offsets to a prefix sum over all parents - // (offsets[i] == offsets[i+1] means parent i has no children) so the consumer skips - // zero-length ranges. Until that scan exists, the owned-copy representation below is the - // contract. + // 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 parentPositions; + std::shared_ptr parentSelVector; std::vector 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(); @@ -103,30 +72,17 @@ class LBUG_API DataChunkState { DASSERT(packedChildSlices.has_value()); return *packedChildSlices; } - void setPackedChildSlices(std::vector parentPositions, std::vector offsets) { - DASSERT(offsets.size() == parentPositions.size() + 1); - packedChildSlices = PackedChildSlices{std::move(parentPositions), std::move(offsets)}; + void setPackedChildSlices(std::shared_ptr parentSelVector, + std::vector offsets) { + DASSERT(parentSelVector != nullptr); + DASSERT(offsets.size() == parentSelVector->getSelSize() + 1); + packedChildSlices = PackedChildSlices{std::move(parentSelVector), std::move(offsets)}; } - void setSingleParentPackedChildSlice(sel_t parentPosition, sel_t numValues) { - setPackedChildSlices({parentPosition}, {0, numValues}); - } - - // 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 parentSelVector, + sel_t numValues) { + DASSERT(parentSelVector->getSelSize() == 1); + setPackedChildSlices(std::move(parentSelVector), {0, numValues}); } void clearPackedChildSlices() { packedChildSlices.reset(); } diff --git a/src/include/processor/operator/scan/scan_rel_table.h b/src/include/processor/operator/scan/scan_rel_table.h index 1ca51571c..07401fe43 100644 --- a/src/include/processor/operator/scan/scan_rel_table.h +++ b/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 copy() override { + std::unique_ptr result; if (sourceMode) { if (sourceNodeScanMode) { - return std::make_unique(opInfo.copy(), tableInfo.copy(), + result = std::make_unique(opInfo.copy(), tableInfo.copy(), copyVector(sourceNodeTableInfos), sourceNodeSharedStates, sourceNodeProgressSharedState, sourceNodeScanInfo.copy(), id, printInfo->copy(), operatorType); + } else { + result = std::make_unique(opInfo.copy(), tableInfo.copy(), + sourceNodeTables, id, printInfo->copy(), operatorType); } - return std::make_unique(opInfo.copy(), tableInfo.copy(), sourceNodeTables, - id, printInfo->copy(), operatorType); + } else { + result = std::make_unique(opInfo.copy(), tableInfo.copy(), + children[0]->copy(), id, printInfo->copy(), operatorType); } - return std::make_unique(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; diff --git a/src/include/storage/table/csr_node_group.h b/src/include/storage/table/csr_node_group.h index 89ae74045..94478abb5 100644 --- a/src/include/storage/table/csr_node_group.h +++ b/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 { diff --git a/src/include/storage/table/rel_table.h b/src/include/storage/table/rel_table.h index fc3851e04..c2a2b6cfe 100644 --- a/src/include/storage/table/rel_table.h +++ b/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 packedChildOffsets; + std::unique_ptr localTableScanState; // Optional state used by Arrow-backed relationship tables. Keep it on the common scan state so diff --git a/src/processor/map/map_aggregate.cpp b/src/processor/map/map_aggregate.cpp index 805d3a73c..75da20ec4 100644 --- a/src/processor/map/map_aggregate.cpp +++ b/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 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(packedChildPhysicalOp.get())) { + scanRelTable->setMultiParentPackedScanEnabled(true); + } auto sink = std::make_unique(sharedState, info, - mapper.mapOperator(packedChild), mapper.getOperatorID(), + std::move(packedChildPhysicalOp), mapper.getOperatorID(), std::make_unique(logicalFilter.getPredicate(), agg.getKeys())); sink->setDescriptor(std::make_unique(packedChildSchema)); diff --git a/src/processor/operator/aggregate/packed_filtered_count.cpp b/src/processor/operator/aggregate/packed_filtered_count.cpp index 955fbc971..5cba49968 100644 --- a/src/processor/operator/aggregate/packed_filtered_count.cpp +++ b/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(parentSelVector[parentIdx]); + uint64_t parentCount = 0; + for (auto rhsIdx = start; rhsIdx < end; ++rhsIdx) { + const auto rhsValue = rhsValueVector->getValue(rhsSelVector[rhsIdx]); + if ((lhsValue + rhsValue) % 10 == 0) { + parentCount += baseMultiplicity; + } + } + if (parentCount > 0) { + localCounts[groupKeyVector->getValue(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(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(groupKeyPos)] += count; } metrics->numOutputTuple.incrementByOne(); diff --git a/src/processor/operator/scan/scan_rel_table.cpp b/src/processor/operator/scan/scan_rel_table.cpp index a77b6bff2..2267a0f67 100644 --- a/src/processor/operator/scan/scan_rel_table.cpp +++ b/src/processor/operator/scan/scan_rel_table.cpp @@ -110,6 +110,7 @@ void ScanRelTable::initLocalStateInternal(ResultSet* resultSet, ExecutionContext scanState = std::make_unique(*MemoryManager::Get(*clientContext), boundNodeIDVector, outVectors, nbrNodeIDVector->state); } + scanState->packedMultiParentScan = multiParentPackedScanEnabled; tableInfo.initScanState(*scanState, outVectors, clientContext); // The native RelTable::initScanState reads from the bound-node // nodeIDVector to pick a node group, so it must run after a child tuple @@ -226,38 +227,26 @@ void ScanRelTable::updatePackedChildSlices(sel_t outputSize) const { scanState->outState->clearPackedChildSlices(); return; } - // See docs/multi_parent_lifetime.md for the representation/lifetime contract of the - // descriptor written here (owned copy; synchronous consumption only). - // - // The CSR scan sets nodeIDVector to flat, pointing its selVector[0] at the actual parent - // whose children are currently materialized in the output vector (see - // RelTableScanState::setNodeIDVectorToFlat). We must use that position as the parent - // position, NOT currBoundNodeIdx, because currBoundNodeIdx may already have advanced past - // this parent by the time we get here (it is incremented when a parent's CSR list is fully - // consumed within a single scan() call). The current scan architecture processes one parent - // per output batch, so we set a single-parent slice (overwriting any previous one). - const auto& boundSelVector = scanState->nodeIDVector->state->getSelVector(); - DASSERT(boundSelVector.getSelSize() == 1); - scanState->outState->setSingleParentPackedChildSlice(boundSelVector[0], outputSize); -} - -void ScanRelTable::reservePackedChildSlicesForBatch() const { - if (operatorType != PhysicalOperatorType::PACKED_EXTEND) { - return; - } - // cachedBoundNodeSelVector holds the bound-node positions for the current input batch and is - // (re)populated by RelTableScanState::initCachedBoundNodeIDSelVector() during initScanState. - // Its selSize is the number of parents that may produce children in this batch. - // - // Only reserve when more than one parent is in flight: the single-parent path uses - // setSingleParentPackedChildSlice (overwrite), which replaces the descriptor and would throw - // away a reservation. Reserving for numParents > 1 keeps the multi-parent append() path - // reallocation-free without adding a wasted allocation to the common single-parent path. - const auto numParents = scanState->cachedBoundNodeSelVector.getSelSize(); - if (numParents <= 1) { - return; + // Attach the PackedChildSlices descriptor to the output (nbr/child) chunk state. See + // docs/multi_parent_lifetime.md for the representation/lifetime contract: the descriptor + // aliases the bound chunk state's selection vector via shared_ptr and is valid only for + // synchronous consumption of this output batch. + const auto boundSelVector = scanState->nodeIDVector->state->getSelVectorShared(); + if (boundSelVector->getSelSize() > 1) { + // Multi-parent packed batch: the CSR scan served the children of several parents and + // recorded the prefix-sum offsets over the served parents in packedChildOffsets (the + // bound vector's chunk state was switched to unflat holding exactly those parents). + DASSERT(!scanState->packedChildOffsets.empty()); + DASSERT(scanState->packedChildOffsets.back() == outputSize); + scanState->outState->setPackedChildSlices(boundSelVector, + std::move(scanState->packedChildOffsets)); + } else { + // Single-parent batch: the bound vector is flat pointing its selVector[0] at the actual + // parent whose children are currently materialized in the output vector (see + // RelTableScanState::setNodeIDVectorToFlat). + DASSERT(boundSelVector->getSelSize() == 1); + scanState->outState->setSingleParentPackedChildSlice(boundSelVector, outputSize); } - scanState->outState->reservePackedChildSlices(numParents); } bool ScanRelTable::getNextTuplesInternal(ExecutionContext* context) { @@ -276,9 +265,6 @@ bool ScanRelTable::getNextTuplesInternal(ExecutionContext* context) { if (!fetchNextBoundNodeBatch(transaction)) { return false; } - // fetchNextBoundNodeBatch established a new input batch (and repopulated - // cachedBoundNodeSelVector via initScanState); reserve for the new parent count. - reservePackedChildSlicesForBatch(); } } while (true) { @@ -295,9 +281,6 @@ bool ScanRelTable::getNextTuplesInternal(ExecutionContext* context) { return false; } tableInfo.table->initScanState(transaction, *scanState); - // A new input batch was just pulled and initScanState repopulated - // cachedBoundNodeSelVector; reserve for the new parent count. - reservePackedChildSlicesForBatch(); } } diff --git a/src/storage/table/csr_node_group.cpp b/src/storage/table/csr_node_group.cpp index 7818a7ee7..35173924c 100644 --- a/src/storage/table/csr_node_group.cpp +++ b/src/storage/table/csr_node_group.cpp @@ -58,6 +58,96 @@ bool CSRNodeGroupScanState::tryScanCachedTuples(RelTableScanState& tableScanStat return true; } +bool CSRNodeGroupScanState::tryScanCachedTuplesPacked(RelTableScanState& tableScanState) { + if (numCachedRows == 0 || + tableScanState.currBoundNodeIdx >= tableScanState.cachedBoundNodeSelVector.getSelSize()) { + return false; + } + auto& outSelVector = tableScanState.outState->getSelVectorUnsafe(); + outSelVector.setToFiltered(); + auto& boundSelVector = tableScanState.nodeIDVector->state->getSelVectorUnsafe(); + boundSelVector.setToFiltered(); + auto& packedChildOffsets = tableScanState.packedChildOffsets; + packedChildOffsets.clear(); + packedChildOffsets.push_back(0); + sel_t numSelected = 0; + sel_t numServedParents = 0; + while (tableScanState.currBoundNodeIdx < tableScanState.cachedBoundNodeSelVector.getSelSize()) { + const auto boundNodePos = + tableScanState.cachedBoundNodeSelVector[tableScanState.currBoundNodeIdx]; + const auto boundNodeOffset = tableScanState.nodeIDVector->readNodeOffset(boundNodePos); + const auto boundNodeOffsetInGroup = boundNodeOffset % StorageConfig::NODE_GROUP_SIZE; + const auto startCSROffset = header->getStartCSROffset(boundNodeOffsetInGroup); + const auto csrLength = header->getCSRLength(boundNodeOffsetInGroup); + if (startCSROffset > nextCachedRowToScan) { + // Jump forward to this parent's list. Parents are visited in CSR order, so this + // only skips over rows of parents already fully consumed. + nextCachedRowToScan = startCSROffset; + } + if (nextCachedRowToScan >= nextRowToScan || + nextCachedRowToScan < nextRowToScan - numCachedRows) { + // This parent's list is outside the cached window. Return the batch accumulated so + // far; the outer scan loop refreshes the cache and resumes with this parent. + break; + } + const auto numRowsToScan = + std::min(nextRowToScan, startCSROffset + csrLength) - nextCachedRowToScan; + const auto numToScan = + std::min(numRowsToScan, DEFAULT_VECTOR_CAPACITY - numSelected); + const auto startCachedRow = nextCachedRowToScan - (nextRowToScan - numCachedRows); + sel_t numSelectedForParent = 0; + if (cachedScannedVectorsSelBitset.has_value()) { + const auto& cachedScannedVectorsSelBitset = *this->cachedScannedVectorsSelBitset; + for (auto i = 0u; i < numToScan; i++) { + const auto rowIdx = startCachedRow + i; + outSelVector[numSelected] = rowIdx; + numSelected += cachedScannedVectorsSelBitset[rowIdx]; + numSelectedForParent += cachedScannedVectorsSelBitset[rowIdx]; + } + } else { + for (auto i = 0u; i < numToScan; i++) { + outSelVector[numSelected++] = startCachedRow + i; + } + numSelectedForParent = numToScan; + } + nextCachedRowToScan += numToScan; + if (numSelectedForParent > 0) { + boundSelVector[numServedParents++] = boundNodePos; + packedChildOffsets.push_back(numSelected); + } + if (numToScan < numRowsToScan) { + // Output capacity reached mid-list (numToScan == 0 with rows remaining also lands + // here): return the batch and let the next one continue with this parent. + break; + } + if ((startCSROffset + csrLength) <= nextCachedRowToScan) { + // Parent's list fully consumed (also covers zero-length lists); move on to the next + // parent within this batch. + tableScanState.currBoundNodeIdx++; + nextCachedRowToScan = 0; + continue; + } + // The parent's list continues beyond the cached window; return the batch and resume + // this parent after the cache is refreshed. + break; + } + if (numServedParents == 0) { + packedChildOffsets.clear(); + return false; + } + if (numServedParents == 1) { + // Preserve the one-parent-per-batch contract when only a single parent was served. + tableScanState.setNodeIDVectorToFlat(boundSelVector[0]); + } else { + tableScanState.nodeIDVector->state->setToUnflat(); + boundSelVector.setSelSize(numServedParents); + } + outSelVector.setSelSize(numSelected); + DASSERT(packedChildOffsets.size() == size_t(numServedParents) + 1); + DASSERT(packedChildOffsets.back() == numSelected); + return true; +} + void CSRNodeGroup::initializeScanState(const Transaction* transaction, TableScanState& state) const { auto& relScanState = state.cast(); @@ -181,7 +271,9 @@ NodeGroupScanResult CSRNodeGroup::scanCommittedPersistent(const Transaction* tra NodeGroupScanResult CSRNodeGroup::scanCommittedPersistentWithCache(const Transaction* transaction, RelTableScanState& tableState, CSRNodeGroupScanState& nodeGroupScanState) const { while (true) { - while (nodeGroupScanState.tryScanCachedTuples(tableState)) { + while (tableState.packedMultiParentScan ? + nodeGroupScanState.tryScanCachedTuplesPacked(tableState) : + nodeGroupScanState.tryScanCachedTuples(tableState)) { if (tableState.outState->getSelVector().getSelSize() > 0) { // Note: This is a dummy return value. return NodeGroupScanResult{nodeGroupScanState.nextRowToScan, diff --git a/test/planner/cardinality_test.cpp b/test/planner/cardinality_test.cpp index efce73eae..0e505d57d 100644 --- a/test/planner/cardinality_test.cpp +++ b/test/planner/cardinality_test.cpp @@ -1,12 +1,28 @@ +#include +#include +#include +#include + +#include "catalog/catalog.h" +#include "catalog/catalog_entry/rel_group_catalog_entry.h" #include "common/data_chunk/data_chunk_state.h" #include "graph_test/private_graph_test.h" +#include "main/client_context.h" #include "planner/operator/logical_plan_util.h" +#include "storage/buffer_manager/memory_manager.h" +#include "storage/storage_manager.h" +#include "storage/table/rel_table.h" +#include "test_helper/test_helper.h" #include "test_runner/test_runner.h" +#include "transaction/transaction.h" #include namespace lbug { namespace testing { +using common::nodeID_t; +using common::sel_t; + class CardinalityTest : public DBTest { public: std::string getInputDir() override { @@ -222,102 +238,256 @@ TEST_F(CardinalityTest, TestPackedExtendDropsParentsWithoutMatches) { EXPECT_EQ(1, foundAIds[0]); } +TEST_F(CardinalityTest, TestMultiParentPackedScan) { + ASSERT_TRUE(conn->query("CREATE NODE TABLE MPerson(id INT64, PRIMARY KEY(id));")->isSuccess()); + ASSERT_TRUE(conn->query("CREATE REL TABLE MKnows(FROM MPerson TO MPerson);")->isSuccess()); + + // 3000 parents; 3/4 of them have 3 children each (6750 edges total, more than one output + // batch of 2048 children), the rest are zero-child parents. The small per-parent degree + // forces the multi-parent packed CSR scan to serve children of many parents per output + // batch (a single-parent batch would hold at most one parent's children). + const common::offset_t numNodes = 3000; + const auto tmpDir = TestHelper::getTempDir("multi-parent-packed-scan"); + const auto nodeCSV = tmpDir + "/nodes.csv"; + const auto relCSV = tmpDir + "/rels.csv"; + std::unordered_map expectedCounts; + { + std::ofstream nodeFile(nodeCSV); + std::ofstream relFile(relCSV); + for (common::offset_t i = 0; i < numNodes; i++) { + nodeFile << i << "\n"; + if (i % 4 == 0) { + continue; // zero-child parent + } + for (common::offset_t j = 1; j <= 3; j++) { + const auto b = (i + j) % numNodes; + relFile << i << "," << b << "\n"; + if ((i + b) % 10 == 0) { + expectedCounts[i]++; + } + } + } + } + ASSERT_TRUE(conn->query(std::format("COPY MPerson FROM '{}'", nodeCSV))->isSuccess()); + ASSERT_TRUE(conn->query(std::format("COPY MKnows FROM '{}'", relCSV))->isSuccess()); + // Materialize the committed data into persistent CSR chunk groups so the rel scan takes + // the persistent with-cache path where multi-parent packing is implemented. + ASSERT_TRUE(conn->query("CHECKPOINT")->isSuccess()); + + // Resolve node offsets: parallel COPY does not assign offsets in CSV order, so ids and + // offsets are permuted relative to each other. + std::unordered_map idToOffset; + { + auto res = conn->query("MATCH (a:MPerson) RETURN a.id, offset(id(a))"); + ASSERT_TRUE(res->isSuccess()) << res->getErrorMessage(); + while (res->hasNext()) { + const auto tup = res->getNext(); + idToOffset[tup->getValue(0)->getValue()] = + tup->getValue(1)->getValue(); + } + ASSERT_EQ(idToOffset.size(), size_t(numNodes)); + } + std::unordered_map> expectedEdges; + for (common::offset_t i = 0; i < numNodes; i++) { + if (i % 4 == 0) { + continue; + } + for (common::offset_t j = 1; j <= 3; j++) { + expectedEdges[idToOffset.at(i)].push_back(idToOffset.at((i + j) % numNodes)); + } + } + + // --- SQL level: the PackedFilteredCount pattern (predicate over an nbr property routes the + // nbr property fetch through a hash join, so these batches stay single-parent — the + // materialization boundary is exactly what multi-parent packing must not cross; see + // docs/multi_parent_lifetime.md). Packed and unpacked plans must agree. + const auto packedCountQuery = + "MATCH (a:MPerson)-[e:MKnows]->(b:MPerson) WHERE (a.id + b.id) % 10 = 0 " + "RETURN a.id, COUNT(*)"; + auto collectCounts = [&](const std::string& query) { + auto res = conn->query(query); + EXPECT_TRUE(res->isSuccess()) << res->getErrorMessage(); + std::unordered_map counts; + while (res->hasNext()) { + const auto tup = res->getNext(); + counts[tup->getValue(0)->getValue()] = tup->getValue(1)->getValue(); + } + return counts; + }; + ASSERT_TRUE(conn->query("CALL enable_packed_path_extend=true")->isSuccess()); + EXPECT_EQ(expectedCounts, collectCounts(packedCountQuery)); + ASSERT_TRUE(conn->query("CALL enable_packed_path_extend=false")->isSuccess()); + EXPECT_EQ(expectedCounts, collectCounts(packedCountQuery)); + + // Standard consumers keep the one-parent-per-batch contract and must see every edge. + ASSERT_TRUE(conn->query("CALL enable_packed_path_extend=true")->isSuccess()); + auto totalRes = conn->query("MATCH (a:MPerson)-[:MKnows]->(b:MPerson) RETURN count(*)"); + ASSERT_TRUE(totalRes->isSuccess()) << totalRes->getErrorMessage(); + ASSERT_TRUE(totalRes->hasNext()); + EXPECT_EQ(6750, totalRes->getNext()->getValue(0)->getValue()); + EXPECT_FALSE(totalRes->hasNext()); + + // --- Storage level: drive the rel scan directly with multi-parent packing enabled and + // verify that batches carry many parents, that the prefix-sum offsets describe each + // parent's children, and that every edge is served exactly once. + ASSERT_TRUE(conn->query("BEGIN TRANSACTION")->isSuccess()); + auto* clientContext = conn->getClientContext(); + auto transaction = transaction::Transaction::Get(*clientContext); + auto* catalog = catalog::Catalog::Get(*clientContext); + auto* nodeEntry = catalog->getTableCatalogEntry(transaction, "MPerson"); + auto* relEntry = catalog->getTableCatalogEntry(transaction, "MKnows"); + const auto nodeTableID = nodeEntry->getTableID(); + const auto relTableID = + relEntry->ptrCast()->getSingleRelEntryInfo().oid; + auto* relTable = storage::StorageManager::Get(*clientContext) + ->getTable(relTableID) + ->ptrCast(); + auto* mm = storage::MemoryManager::Get(*clientContext); + + auto boundState = std::make_shared(); + auto boundVector = + std::make_unique(common::LogicalType::INTERNAL_ID(), mm); + boundVector->state = boundState; + auto outState = std::make_shared(); + auto nbrVector = std::make_unique(common::LogicalType::INTERNAL_ID(), mm); + nbrVector->state = outState; + + storage::RelTableScanState scanState(*mm, boundVector.get(), {nbrVector.get()}, outState); + scanState.setToTable(transaction, relTable, {0}, {}, common::RelDataDirection::FWD); + scanState.packedMultiParentScan = true; + + common::offset_t nextParent = 0; + auto feedNextParentBatch = [&]() { + if (nextParent >= numNodes) { + return false; + } + const auto end = std::min(nextParent + common::DEFAULT_VECTOR_CAPACITY, numNodes); + const auto count = end - nextParent; + boundState->setToUnflat(); + boundState->getSelVectorUnsafe().setToUnfiltered(count); + for (common::offset_t i = 0; i < count; i++) { + boundVector->setValue(i, nodeID_t{nextParent + i, nodeTableID}); + } + nextParent = end; + relTable->initScanState(transaction, scanState); + return true; + }; + + std::unordered_map> edges; + uint64_t numBatches = 0; + uint64_t numMultiParentBatches = 0; + ASSERT_TRUE(feedNextParentBatch()); + for (;;) { + if (!relTable->scan(transaction, scanState)) { + if (!feedNextParentBatch()) { + break; + } + continue; + } + numBatches++; + const auto& boundSelVector = scanState.nodeIDVector->state->getSelVector(); + const auto numParentsInBatch = boundSelVector.getSelSize(); + if (numParentsInBatch > 1) { + numMultiParentBatches++; + } + const auto& outSelVector = scanState.outState->getSelVector(); + const auto& offsets = scanState.packedChildOffsets; + if (numParentsInBatch > 1) { + ASSERT_EQ(offsets.size(), size_t(numParentsInBatch) + 1); + ASSERT_EQ(offsets.back(), outSelVector.getSelSize()); + } + for (sel_t p = 0; p < numParentsInBatch; p++) { + const auto parentOffset = boundVector->getValue(boundSelVector[p]).offset; + const auto start = numParentsInBatch > 1 ? offsets[p] : 0; + const auto end = numParentsInBatch > 1 ? offsets[p + 1] : outSelVector.getSelSize(); + for (auto i = start; i < end; i++) { + edges[parentOffset].push_back( + nbrVector->getValue(outSelVector[i]).offset); + } + } + } + // With degree 3 and 2048-child output batches, packing must actually engage: without it, + // every batch would hold the children of exactly one parent. + ASSERT_GT(numMultiParentBatches, 0u); + ASSERT_EQ(edges.size(), expectedEdges.size()); + for (auto& [parent, children] : expectedEdges) { + auto actual = edges.at(parent); + auto expected = children; + std::sort(actual.begin(), actual.end()); + std::sort(expected.begin(), expected.end()); + ASSERT_EQ(actual, expected); + } + conn->query("COMMIT"); +} + TEST_F(CardinalityTest, TestPackedChildSliceState) { common::DataChunkState state; EXPECT_FALSE(state.hasPackedChildSlices()); - state.setSingleParentPackedChildSlice(3, 7); + // Single-parent convenience: the parent selection vector holds exactly one position. + auto singleParentSel = std::make_shared(1); + singleParentSel->setToFiltered(1); + (*singleParentSel)[0] = 3; + state.setSingleParentPackedChildSlice(singleParentSel, 7); ASSERT_TRUE(state.hasPackedChildSlices()); const auto& singleParentSlices = state.getPackedChildSlices(); ASSERT_EQ(1, singleParentSlices.getNumParents()); - EXPECT_EQ(3, singleParentSlices.parentPositions[0]); + EXPECT_EQ(3, (*singleParentSlices.parentSelVector)[0]); EXPECT_EQ(0, singleParentSlices.offsets[0]); EXPECT_EQ(7, singleParentSlices.offsets[1]); EXPECT_EQ(7, singleParentSlices.getNumValues()); - state.setPackedChildSlices({1, 4}, {0, 2, 5}); + // Multi-parent: prefix-sum offsets over ALL parents, zeros allowed. Parent 0 owns output + // positions [0,2), parent 1 has no children ([2,2)), parent 2 owns [2,5), parent 3 [5,7). + auto multiParentSel = std::make_shared(4); + multiParentSel->setToFiltered(4); + for (auto i = 0u; i < 4; i++) { + (*multiParentSel)[i] = i + 1; + } + state.setPackedChildSlices(multiParentSel, {0, 2, 2, 5, 7}); const auto& multiParentSlices = state.getPackedChildSlices(); - ASSERT_EQ(2, multiParentSlices.getNumParents()); - EXPECT_EQ(1, multiParentSlices.parentPositions[0]); - EXPECT_EQ(4, multiParentSlices.parentPositions[1]); - EXPECT_EQ(5, multiParentSlices.getNumValues()); + ASSERT_EQ(4, multiParentSlices.getNumParents()); + EXPECT_EQ(7, multiParentSlices.getNumValues()); + EXPECT_EQ(2, multiParentSlices.offsets[1] - multiParentSlices.offsets[0]); + EXPECT_EQ(0, multiParentSlices.offsets[2] - multiParentSlices.offsets[1]); + EXPECT_EQ(3, multiParentSlices.offsets[3] - multiParentSlices.offsets[2]); + EXPECT_EQ(2, multiParentSlices.offsets[4] - multiParentSlices.offsets[3]); state.clearPackedChildSlices(); EXPECT_FALSE(state.hasPackedChildSlices()); } -TEST_F(CardinalityTest, TestPackedChildSliceAppend) { - common::DataChunkState state; - EXPECT_FALSE(state.hasPackedChildSlices()); - - // Append first parent - state.appendPackedChildSlice(2, 3); - ASSERT_TRUE(state.hasPackedChildSlices()); - { - const auto& slices = state.getPackedChildSlices(); - ASSERT_EQ(1, slices.getNumParents()); - EXPECT_EQ(2, slices.parentPositions[0]); - ASSERT_EQ(2, slices.offsets.size()); - EXPECT_EQ(0, slices.offsets[0]); - EXPECT_EQ(3, slices.offsets[1]); - EXPECT_EQ(3, slices.getNumValues()); - } - - // Append second parent - state.appendPackedChildSlice(5, 4); - { - const auto& slices = state.getPackedChildSlices(); - ASSERT_EQ(2, slices.getNumParents()); - EXPECT_EQ(2, slices.parentPositions[0]); - EXPECT_EQ(5, slices.parentPositions[1]); - ASSERT_EQ(3, slices.offsets.size()); - EXPECT_EQ(0, slices.offsets[0]); - EXPECT_EQ(3, slices.offsets[1]); - EXPECT_EQ(7, slices.offsets[2]); - EXPECT_EQ(7, slices.getNumValues()); - } - - // Append zero-sized parent should still extend offsets correctly - state.appendPackedChildSlice(7, 0); - { - const auto& slices = state.getPackedChildSlices(); - ASSERT_EQ(3, slices.getNumParents()); - EXPECT_EQ(7, slices.parentPositions[2]); - ASSERT_EQ(4, slices.offsets.size()); - EXPECT_EQ(7, slices.offsets[2]); - EXPECT_EQ(7, slices.offsets[3]); - EXPECT_EQ(7, slices.getNumValues()); - } -} - -TEST_F(CardinalityTest, TestPackedChildSliceOwnedCopyLifetime) { +TEST_F(CardinalityTest, TestPackedChildSliceAliasedParentSelection) { common::DataChunkState state; - // Simulate the CSR scan: the bound-node (parent) selection vector is flat with selSize 1, - // pointing its position 0 at parent row 9. auto& selVector = state.getSelVectorUnsafe(); + // Simulate the scan: the bound-node (parent) selection vector is filtered with selSize 1, + // pointing its position 0 at parent row 9. selVector.setToFiltered(1); selVector[0] = 9; - state.setSingleParentPackedChildSlice(selVector[0], 5); - { - const auto& slices = state.getPackedChildSlices(); - ASSERT_EQ(1, slices.getNumParents()); - EXPECT_EQ(9, slices.parentPositions[0]); - EXPECT_EQ(5, slices.getNumValues()); - } - - // The selection vector's contents are mutable in place (setToFiltered/setToUnfiltered - // rewrite the buffer). Rewriting it here is what the next input batch would do. Since - // parentPositions is an OWNED copy — not an alias into the selection vector — the - // descriptor must keep the parent position recorded for this batch. See - // docs/multi_parent_lifetime.md for the synchronous-consumption lifetime rule. + // The descriptor ALIASES the parent chunk state's selection vector (shared_ptr keeps the + // object alive; contents are not snapshotted). + state.setSingleParentPackedChildSlice(state.getSelVectorShared(), 5); + ASSERT_TRUE(state.hasPackedChildSlices()); + const auto& slices = state.getPackedChildSlices(); + ASSERT_EQ(1, slices.getNumParents()); + EXPECT_EQ(9, (*slices.parentSelVector)[0]); + + // The bound chunk's selection vector contents are rewritten in place for the next input + // batch (setToFiltered/setToUnfiltered). The alias observes the rewrite: consumers must + // finish reading the descriptor synchronously with the output batch. See + // docs/multi_parent_lifetime.md for the lifetime rule. selVector[0] = 42; - selVector.setToUnfiltered(1); - { - const auto& slices = state.getPackedChildSlices(); - EXPECT_EQ(9, slices.parentPositions[0]); - EXPECT_EQ(5, slices.getNumValues()); - } + EXPECT_EQ(42, (*slices.parentSelVector)[0]); + + // The shared_ptr keeps the SelectionVector object alive even if the chunk state replaces + // its selection vector. + auto replacement = std::make_shared(1); + replacement->setToFiltered(1); + (*replacement)[0] = 7; + state.setSelVector(replacement); + EXPECT_EQ(42, (*slices.parentSelVector)[0]); + EXPECT_EQ(5, slices.getNumValues()); } } // namespace testing From b93525909c8897b7af62f628b6066cc2c5395802 Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sun, 30 Aug 2026 21:52:12 -0700 Subject: [PATCH 4/4] Allow flat vectors in FactorizedTable unflat column append Debug builds assert in FactorizedTable::copyVectorToUnflatColumn / appendVectorToUnflatTupleBlocks when a flat vector is appended to an unflat column. The implementation already handles this correctly: a flat vector (single selected value) is stored as a one-element unflat overflow value, which is exactly the right representation. This shape is reached by the packed filtered count pipeline: the packed extend keeps its bound node group unflat at plan time (getGroupsPosToFlatten is empty, so no Flatten operators are inserted), while the CSR scan presents the bound chunk flat with selSize 1 per output batch. When the acc-hash-join SIP rewrite places an Accumulate (FactorizedTable writer) directly above the packed extend, the append hits the over-strict assert on DASSERT-enabled (Debug) builds. Release builds run the same path and produce correct results. Relax the asserts to require only that a flat vector carry a single selected value, matching the implemented semantics. --- src/processor/result/factorized_table.cpp | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/processor/result/factorized_table.cpp b/src/processor/result/factorized_table.cpp index cc2fd521c..e6809b34a 100644 --- a/src/processor/result/factorized_table.cpp +++ b/src/processor/result/factorized_table.cpp @@ -485,7 +485,10 @@ void FactorizedTable::copyUnflatVectorToFlatColumn(const ValueVector& vector, // factorizedTable. NullMasks are stored inside the overflow buffer. void FactorizedTable::copyVectorToUnflatColumn(const ValueVector& vector, const BlockAppendingInfo& blockAppendInfo, ft_col_idx_t colIdx) { - DASSERT(!vector.state->isFlat()); + // A flat vector is stored as a one-element unflat value. This happens when a group is + // unflat at plan time but flat at run time, e.g. the packed extend's bound node chunk + // (flat with selSize 1 per output batch) appended by an Accumulate directly above it. + DASSERT(!vector.state->isFlat() || vector.state->getSelVector().getSelSize() == 1); auto unflatTupleValue = appendVectorToUnflatTupleBlocks(vector, colIdx); auto blockPtr = blockAppendInfo.data + tableSchema.getColOffset(colIdx); for (auto i = 0u; i < blockAppendInfo.numTuplesToAppend; i++) { @@ -505,7 +508,9 @@ void FactorizedTable::copyVectorToColumn(const ValueVector& vector, overflow_value_t FactorizedTable::appendVectorToUnflatTupleBlocks(const ValueVector& vector, ft_col_idx_t colIdx) { - DASSERT(!vector.state->isFlat()); + // See copyVectorToUnflatColumn: flat vectors (single selected value) are allowed and are + // stored as a one-element unflat value. + DASSERT(!vector.state->isFlat() || vector.state->getSelVector().getSelSize() == 1); auto numFlatTuplesInVector = vector.state->getSelVector().getSelSize(); auto numBytesPerValue = LogicalTypeUtils::getRowLayoutSize(vector.dataType); auto numBytesForData = numBytesPerValue * numFlatTuplesInVector;