| 630 | } |
| 631 | |
| 632 | Status DataProcessorService::removeTransform(std::string_view namespaced_key) { |
| 633 | const auto it = transform_recipes_.find(std::string(namespaced_key)); |
| 634 | if (it == transform_recipes_.end()) { |
| 635 | return PJ::unexpected("pj.data_processors: unknown transform '" + std::string(namespaced_key) + "'"); |
| 636 | } |
| 637 | const NodeId node = it->second.node_id; |
| 638 | const std::vector<TopicId> outputs = it->second.output_topic_ids; |
| 639 | transform_recipes_.erase(it); |
| 640 | (void)derived_->removeNode(node); |
| 641 | // Retire the materialized output so the catalog drops it (removeNode alone leaves a zombie). |
| 642 | for (const TopicId out : outputs) { |
| 643 | engine_.retireTopic(out); |
| 644 | } |
| 645 | return PJ::okStatus(); |
| 646 | } |
| 647 | |
| 648 | void DataProcessorService::clearTransformsForPlugin(std::string_view plugin_id) { |
| 649 | std::vector<std::string> keys; |