Method mapExtend

src/processor/map/map_extend.cpp:164–287

}

std::unique_ptr<PhysicalOperator> PlanMapper::mapExtend(const LogicalOperator* logicalOperator) {
    auto extend = logicalOperator->constPtrCast<LogicalExtend>();
    auto outFSchema = extend->getSchema();
    auto inFSchema = extend->getChild(0)->getSchema();
    auto boundNode = extend->getBoundNode();
    auto nbrNode = extend->getNbrNode();
    auto rel = extend->getRel();
    auto extendDirection = extend->getDirection();
    const auto physicalOperatorType =
        logicalOperator->getOperatorType() == LogicalOperatorType::PACKED_EXTEND ?
            PhysicalOperatorType::PACKED_EXTEND :
            PhysicalOperatorType::SCAN_REL_TABLE;
    auto prevOperator = mapOperator(logicalOperator->getChild(0).get());
    auto inNodeIDPos = getDataPos(*boundNode->getInternalID(), *inFSchema);
    std::vector<DataPos> outVectorsPos;
    auto outNodeIDPos = getDataPos(*nbrNode->getInternalID(), *outFSchema);
    outVectorsPos.push_back(outNodeIDPos);
    for (auto& expression : extend->getProperties()) {
        outVectorsPos.push_back(getDataPos(*expression, *outFSchema));
    }
    auto scanInfo = ScanOpInfo(inNodeIDPos, outVectorsPos);
    std::vector<std::string> tableNames;
    auto storageManager = StorageManager::Get(*clientContext);
    for (auto entry : rel->getEntries()) {
        tableNames.push_back(entry->getName());
    }
    auto printInfo = std::make_unique<ScanRelTablePrintInfo>(tableNames, extend->getProperties(),
        boundNode, rel, nbrNode, extendDirection, rel->getVariableName());
    if (scanSingleRelTable(*rel, *boundNode, extendDirection)) {
        DASSERT(rel->getNumEntries() == 1);
        auto entry = rel->getEntry(0)->ptrCast<RelGroupCatalogEntry>();
        auto relDataDirection = ExtendDirectionUtil::getRelDataDirection(extendDirection);
        auto entryInfo = entry->getSingleRelEntryInfo();
        auto relTable = storageManager->getTable(entryInfo.oid)->ptrCast<RelTable>();
        auto scanRelInfo =
            getRelTableScanInfo(*entry, relDataDirection, relTable, extend->shouldScanNbrID(),
                extend->getProperties(), extend->getPropertyPredicates(), clientContext);
        if (logicalOperator->getChild(0)->getOperatorType() ==
            LogicalOperatorType::SCAN_NODE_TABLE) {
            auto* scanNode = logicalOperator->getChild(0)->ptrCast<LogicalScanNodeTable>();
            if (scanNode->getScanType() == LogicalScanNodeTableType::SCAN) {
                std::vector<NodeTable*> sourceNodeTables;
                std::vector<ScanNodeTableInfo> sourceNodeTableInfos;
                std::vector<std::shared_ptr<ScanNodeTableSharedState>> sourceNodeSharedStates;
                auto expectedBoundTableID = relDataDirection == RelDataDirection::FWD ?
                                                relTable->getFromNodeTableID() :
                                                relTable->getToNodeTableID();
                auto catalog = catalog::Catalog::Get(*clientContext);
                auto transaction = transaction::Transaction::Get(*clientContext);
                for (auto tableID : scanNode->getTableIDs()) {
                    if (tableID == expectedBoundTableID) {
                        auto table = storageManager->getTable(tableID)->ptrCast<NodeTable>();
                        sourceNodeTables.push_back(table);
                        auto tableEntry = catalog->getTableCatalogEntry(transaction, tableID);
                        sourceNodeTableInfos.push_back(
                            getNodeTableScanInfo(*scanNode, table, tableEntry, clientContext));
                        auto semiMask =
                            SemiMaskUtil::createMask(table->getNumTotalRows(transaction));