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 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 6ecbafc38..43c683587 100644 --- a/src/include/common/data_chunk/data_chunk_state.h +++ b/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 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(); @@ -79,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 setSingleParentPackedChildSlice(sel_t parentPosition, sel_t numValues) { - setPackedChildSlices({parentPosition}, {0, numValues}); + 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)}; } - - // 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 52b003cf2..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,35 +227,26 @@ void ScanRelTable::updatePackedChildSlices(sel_t outputSize) const { scanState->outState->clearPackedChildSlices(); return; } - // 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) { @@ -273,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) { @@ -292,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/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; 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 76b4c60fc..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,72 +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) { +TEST_F(CardinalityTest, TestPackedChildSliceAliasedParentSelection) { common::DataChunkState state; - EXPECT_FALSE(state.hasPackedChildSlices()); + 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; - // Append first parent - state.appendPackedChildSlice(2, 3); + // 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(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()); - } + const auto& slices = state.getPackedChildSlices(); + ASSERT_EQ(1, slices.getNumParents()); + EXPECT_EQ(9, (*slices.parentSelVector)[0]); - // 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()); - } + // 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; + EXPECT_EQ(42, (*slices.parentSelVector)[0]); - // 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()); - } + // 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