Method installTransform

pj_runtime/src/DataProcessorService.cpp:489–614

}  // namespace

Expected<DataProcessorService::TransformRecipe> DataProcessorService::installTransform(TransformRecipe recipe) {
  // 1. Shape: N inputs -> M outputs (SISO is the 1->1 case). Field-level binding
  //    (a column within a multi-column input) is still a follow-up: inputs resolve
  //    to whole scalar topics here.
  if (recipe.inputs.empty() || recipe.outputs.empty()) {
    return PJ::unexpected("pj.data_processors: transform requires at least one input and one output");
  }
  for (const std::string& out_name : recipe.outputs) {
    if (out_name.empty()) {
      return PJ::unexpected("pj.data_processors: transform output name must be non-empty");
    }
  }
  // 2. Backend (also the missing/unavailable-backend diagnostic on restore).
  auto backend = inferTransformBackend(recipe.script);
  if (!backend.has_value()) {
    return PJ::unexpected(backend.error());
  }
  recipe.backend = std::move(backend).value();
  // 3. Output-name collision, per output (excludes this transform's own outputs, so a replace is allowed).
  for (const std::string& out_name : recipe.outputs) {
    if (outputNameInUse(out_name, recipe.key)) {
      return PJ::unexpected("pj.data_processors: output topic name already in use: " + out_name);
    }
  }
  // 4. Resolve every input by name; the output topics are created in the first input's dataset.
  std::vector<TopicId> input_topic_ids;
  std::vector<std::size_t> input_columns;  // leaf column per input (field-level binding)
  input_topic_ids.reserve(recipe.inputs.size());
  input_columns.reserve(recipe.inputs.size());
  for (const std::string& in_name : recipe.inputs) {
    const auto resolved = resolveInputField(in_name);
    if (!resolved.has_value()) {
      return PJ::unexpected("pj.data_processors: input topic/field not found: " + in_name);
    }
    if (input_topic_ids.empty()) {
      recipe.dataset_id = resolved->dataset_id;
    }
    input_topic_ids.push_back(resolved->topic_id);
    input_columns.push_back(resolved->column);
  }
  // Grouped output: a single structured "topic:f1,f2,..." entry collapses into ONE
  // multi-column topic (parsed host-side, explicit — no prefix guessing). A grouped
  // output is MIMO even when it is the only output entry, so it must not be mistaken
  // for SISO, and its arity (M) is the count of declared fields.
  const auto [mimo_group, mimo_fields] = parseGroupedOutput(recipe.outputs);
  const bool grouped = !mimo_group.empty();
  const bool is_siso = recipe.inputs.size() == 1 && recipe.outputs.size() == 1 && !grouped;
  const std::size_t mimo_num_outputs = grouped ? mimo_fields.size() : recipe.outputs.size();
  // 5. Compile BEFORE tearing down any old node, so a bad script never destroys a
  //    working transform. Use the catalogue's engine when installed, else a transient
  //    one (the built transform shares ownership of the engine, so it outlives it).
  std::shared_ptr<proc::DataProcessor> siso_op;          // set iff is_siso
  std::unique_ptr<scripting::LuaMimoTransform> mimo_op;  // set iff !is_siso
  // Python transforms compile through a transient Python catalogue (the embedded
  // interpreter is process-global). Luau reuses the installed catalogue.
  const bool is_python = recipe.backend == "python";
  if (is_siso) {
    auto compiled = is_python ? scripting::FilterCatalogue(scripting::makePythonEngine())