为文档关联构建哈希表，并合并数组键的匹配结果

1399        // Plain $lookup without $unwind: use HashLookupStage.1400        // Aggregator to assemble the matched foreign documents into an array.1401        SbBlockAggExpr agg{SbExpr{} /*init*/,1402                           SbExpr{} /*blockAgg*/,1403                           b.makeFunction(sbe::EFn::kAddToArray, foreignRecordSlot) /*agg*/};14041405        auto [hl, lookupStageOutputSlot] = b.makeHashLookup(std::move(localKeysSetStage),1406                                                            std::move(foreignKeyStage),1407                                                            localKeysSetSlot,1408                                                            foreignKeySlot,1409                                                            foreignRecordSlot,1410                                                            std::move(agg),1411                                                            boost::none /* optOutputSlot */,1412                                                            state.getCollatorSlot());14131414        // Add a projection that returns an empty array in the "as" field if no foreign row matched.1415        auto [emptyArrayTag, emptyArrayVal] = sbe::value::makeNewArray();1416        auto emptyArrayConstant = b.makeConstant(emptyArrayTag, emptyArrayVal);14171418        SbExpr innerResultProjection =1419            b.makeFillEmpty(lookupStageOutputSlot, std::move(emptyArrayConstant));

150    // Insert the inner side into the hash table.151    innerChild()->open(false);152    value::FixedSizeRow<1 /*N*/> value{1};153    while (innerChild()->getNext() == PlanState::ADVANCED) {154        // Copy the projected value.155        value.reset(0, _inInnerProjectAccessor->getCopyOfValue());156157        // This where we put the value in here. This can grow need to spill.158        size_t bufferIndex = _hashTable.bufferValueOrSpill(value);159        _memoryTracker.value().set(_hashTable.getMemUsage());160161        auto [tagKeyView, valKeyView] = _inInnerMatchAccessor->getViewOfValue();162        if (value::isArray(tagKeyView)) {163            value::ArrayAccessor arrayAccessor;164            arrayAccessor.reset(_inInnerMatchAccessor);165166            while (!arrayAccessor.atEnd()) {167                _hashTable.addHashTableEntry(&arrayAccessor, bufferIndex);168                _memoryTracker.value().set(_hashTable.getMemUsage());169                arrayAccessor.advance();170            }

188    } else {189        // The key is already present in '_memoryHt' so the memory will only grow by one size_t. If190        // we reach the memory limit, the key/value in '_memoryHt' will be evicted from memory and191        // spilled to '_recordStoreHt' along with the new index.192        htIt->second.push_back(valueIndex);193        _computedTotalMemUsage += sizeof(size_t);194        if (_computedTotalMemUsage > _memoryUseInBytesBeforeSpill) {195            if (!hasSpilledHtToDisk()) {196                makeInternalRecordStore();197            }198199            _computedTotalMemUsage -= size_estimator::estimate(keyView.tag, keyView.value);200201            // Evict the hash table value.202            _computedTotalMemUsage -= htIt->second.size() * sizeof(size_t);203            spillIndicesToRecordStore(_recordStoreHt.get(), keyView, htIt->second);204            _memoryHt->erase(htIt);205        }

192PlanState HashLookupStage::getNext() {193    auto optTimer(getOptTimer(_opCtx));194195    PlanState state = outerChild()->getNext();196    if (state == PlanState::ADVANCED) {197        // We just got this outer doc, so reset the $lookup "as" result array accumulator to nothing198        // and the hash table iterator to the outer key.199        _lookupStageOutput.reset(0, value::TagValueView::nothing());200        _hashTable.htIter.reset(_inOuterMatchAccessor->getViewOfValue());201202        // Accumulate all the matching inner docs for the outer key(s).203        accumulateFromValueIndices(_hashTable.htIter.getAllMatchingIndices());204    }205    return trackPlanState(state);206}  // HashLookupStage::getNext

24    value::ArrayEnumerator enumerator(_outerKey.tag, _outerKey.value);25    while (!enumerator.atEnd()) {26        auto elemView = enumerator.getViewOfValue();2728        auto hashTableMatchIter = _hashTable._memoryHt->find(elemView);29        if (hashTableMatchIter != _hashTable._memoryHt->end()) {30            _hashTableMatchVector.insert(_hashTableMatchVector.end(),31                                         hashTableMatchIter->second.begin(),32                                         hashTableMatchIter->second.end());33        } else if (_hashTable._recordStoreHt) {34            // The key wasn't in memory. Check the '_hashTable._recordStoreHt' disk spill.35            auto elemColl = _hashTable.normalizeStringIfCollator(elemView);3637            auto indicesFromRS = _hashTable.readIndicesFromRecordStore(38                _hashTable._recordStoreHt.get(), elemColl.view());39            if (indicesFromRS) {40                _hashTableMatchVector.insert(41                    _hashTableMatchVector.end(), indicesFromRS->begin(), indicesFromRS->end());42            }43        }

47    // Different array elements (or the memory and disk portions of the hash table) may map to the48    // same inner buffer index, so duplicates must be removed.49    std::sort(_hashTableMatchVector.begin(), _hashTableMatchVector.end());50    _hashTableMatchVector.erase(51        std::unique(_hashTableMatchVector.begin(), _hashTableMatchVector.end()),52        _hashTableMatchVector.end());5354    _hashTableMatchVectorIdx = 0;55    _hashTableSearched = true;56}  // LookupHashTableIter::initSearchArray

181void HashLookupStage::accumulateFromValueIndices(const RecordIndexCollection* bufferIndices) {182    for (const size_t bufferIdx : *bufferIndices) {183        boost::optional<value::TagValueView> innerMatch = _hashTable.getValueAtIndex(bufferIdx);184        tassert(10801300, "Expected non-empty innerMatch", innerMatch);185        _outInnerProjectAccessor.reset(*innerMatch);186187        // Run the VM code to "accumulate" the current inner doc into the lookup output array.188        _lookupStageOutput.reset(0 /* column */, _bytecode.run(_aggCode.get()));189    }190}  // HashLookupStage::accumulateFromValueIndices