Method mapCopyRelFrom

src/processor/map/map_copy_from.cpp:121–165

}

std::unique_ptr<PhysicalOperator> PlanMapper::mapCopyRelFrom(
    const LogicalOperator* logicalOperator) {
    auto& copyFrom = logicalOperator->constCast<LogicalCopyFrom>();
    const auto copyFromInfo = copyFrom.getInfo();
    auto partitioner = mapOperator(copyFrom.getChild(0).get());
    DASSERT(partitioner->getOperatorType() == PhysicalOperatorType::PARTITIONER);
    auto partitionerSharedState = partitioner->ptrCast<Partitioner>()->getSharedState();
    const auto catalog = Catalog::Get(*clientContext);
    const auto transaction = transaction::Transaction::Get(*clientContext);
    auto extraInfo = copyFromInfo->extraInfo->constCast<ExtraBoundCopyRelInfo>();
    auto fromTableID =
        catalog->getTableCatalogEntry(transaction, extraInfo.fromTableName)->getTableID();
    auto toTableID =
        catalog->getTableCatalogEntry(transaction, extraInfo.toTableName)->getTableID();
    std::vector<LogicalType> warningColumnTypes;
    for (auto& column : copyFromInfo->getWarningColumns()) {
        warningColumnTypes.push_back(column->getDataType().copy());
    }
    auto fTable =
        FactorizedTableUtils::getSingleStringColumnFTable(MemoryManager::Get(*clientContext));
    auto batchInsertSharedState = std::make_shared<BatchInsertSharedState>(fTable);
    // If the table entry doesn't exist, assume both directions
    std::vector directions = {RelDataDirection::FWD, RelDataDirection::BWD};
    if (catalog->containsTable(transaction, copyFromInfo->tableName)) {
        const auto& relGroupEntry =
            catalog->getTableCatalogEntry(transaction, copyFromInfo->tableName)
                ->constCast<RelGroupCatalogEntry>();
        directions = relGroupEntry.getRelDataDirections();
    }

    auto sink = std::make_unique<DummySimpleSink>(fTable, getOperatorID());
    for (auto direction : directions) {
        auto insertInfo = std::make_unique<RelBatchInsertInfo>(copyFromInfo->tableName,
            copyVector(warningColumnTypes), fromTableID, toTableID, direction);
        auto printInfo = std::make_unique<RelBatchInsertPrintInfo>(copyFromInfo->tableName);
        auto progress = std::make_shared<RelBatchInsertProgressSharedState>();
        auto batchInsert = std::make_unique<RelBatchInsert>(std::move(insertInfo),
            partitionerSharedState, batchInsertSharedState, getOperatorID(), std::move(printInfo),
            progress, std::make_unique<CopyRelBatchInsert>());
        batchInsert->setDescriptor(std::make_unique<ResultSetDescriptor>(copyFrom.getSchema()));
        sink->addChild(std::move(batchInsert));
    }
    sink->addChild(std::move(partitioner));
    return sink;
}

} // namespace processor
} // namespace lbug