DPL: modernize workflow construction code using ranges (#14907)

aalkin · web-flow · commit 70b10405ca2f · 2025-12-16T22:02:47.000+01:00
diff --git a/Framework/Core/include/Framework/AnalysisSupportHelpers.h b/Framework/Core/include/Framework/AnalysisSupportHelpers.h
@@ -39,11 +39,6 @@ struct AnalysisSupportHelpers {
                                          std::vector<InputSpec> const& requestedSpecials,
                                          std::vector<InputSpec>& requestedAODs,
                                          DataProcessorSpec& publisher);
-  static void addMissingOutputsToAnalysisCCDBFetcher(std::vector<OutputSpec> const& providedSpecials,
-                                                     std::vector<InputSpec> const& requestedSpecials,
-                                                     std::vector<InputSpec>& requestedAODs,
-                                                     std::vector<InputSpec>& requestedDYNs,
-                                                     DataProcessorSpec& publisher);
   static void addMissingOutputsToBuilder(std::vector<InputSpec> const& requestedSpecials,
                                          std::vector<InputSpec>& requestedAODs,
                                          std::vector<InputSpec>& requestedDYNs,
diff --git a/Framework/Core/include/Framework/DataSpecViews.h b/Framework/Core/include/Framework/DataSpecViews.h
@@ -31,6 +31,32 @@ static auto filter_not_matching(auto const& provided)
   return std::views::filter([&provided](auto const& input) { return std::none_of(provided.begin(), provided.end(), [&input](auto const& output) { return DataSpecUtils::match(input, output); }); });
 }
 
+static auto filter_matching(auto const& provided)
+{
+  return std::views::filter([&provided](auto const& input) { return std::any_of(provided.begin(), provided.end(), [&input](auto const& output) { return DataSpecUtils::match(input, output); }); });
+}
+
+static auto filter_string_params_with(std::string match)
+{
+  return std::views::filter([match](auto const& param) {
+    return (param.type == VariantType::String) && (param.name.find(match) != std::string::npos);
+  });
+}
+
+static auto input_to_output_specs()
+{
+  return std::views::transform([](auto const& input) {
+    auto concrete = DataSpecUtils::asConcreteDataMatcher(input);
+    return OutputSpec{concrete.origin, concrete.description, concrete.subSpec, input.lifetime, input.metadata};
+  });
+}
+
+static auto params_to_input_specs()
+{
+  return std::views::transform([](auto const& param) {
+    return DataSpecUtils::fromMetadataString(param.defaultValue.template get<std::string>());
+  });
+}
 } // namespace o2::framework::views
 //
 namespace o2::framework::sinks
@@ -54,14 +80,29 @@ struct update_input_list {
   template <std::ranges::input_range R>
   friend Container& operator|(R&& r, update_input_list self)
   {
-    for (auto& item : r) {
+    for (auto const& item : r) {
       auto copy = item;
       DataSpecUtils::updateInputList(self.c, std::move(copy));
     }
     return self.c;
   }
 };
 
+template <class Container>
+struct update_output_list {
+  Container& c;
+  // ends the pipeline, returns the container
+  template <std::ranges::input_range R>
+  friend Container& operator|(R&& r, update_output_list self)
+  {
+    for (auto const& item : r) {
+      auto copy = item;
+      DataSpecUtils::updateOutputList(self.c, std::move(copy));
+    }
+    return self.c;
+  }
+};
+
 } // namespace o2::framework::sinks
 
 #endif // O2_FRAMEWORK_DATASPECVIEWS_H_
diff --git a/Framework/Core/src/AnalysisSupportHelpers.cxx b/Framework/Core/src/AnalysisSupportHelpers.cxx
@@ -11,10 +11,7 @@
 
 #include "Framework/AnalysisSupportHelpers.h"
 #include "Framework/DataOutputDirector.h"
-#include "Framework/OutputObjHeader.h"
-#include "Framework/ControlService.h"
-#include "Framework/EndOfStreamContext.h"
-#include "Framework/DeviceSpec.h"
+#include "Framework/DataSpecViews.h"
 #include "Framework/PluginManager.h"
 #include "Framework/ConfigContext.h"
 #include "WorkflowHelpers.h"
@@ -129,109 +126,59 @@ void AnalysisSupportHelpers::addMissingOutputsToReader(std::vector<OutputSpec> c
                                                        std::vector<InputSpec> const& requestedInputs,
                                                        DataProcessorSpec& publisher)
 {
-  auto matchingOutputFor = [](InputSpec const& requested) {
-    return [&requested](OutputSpec const& provided) {
-      return DataSpecUtils::match(requested, provided);
-    };
-  };
-  for (InputSpec const& requested : requestedInputs) {
-    auto provided = std::find_if(providedOutputs.begin(),
-                                 providedOutputs.end(),
-                                 matchingOutputFor(requested));
-
-    if (provided != providedOutputs.end()) {
-      continue;
-    }
-
-    auto inList = std::find_if(publisher.outputs.begin(),
-                               publisher.outputs.end(),
-                               matchingOutputFor(requested));
-    if (inList != publisher.outputs.end()) {
-      continue;
-    }
-
-    auto concrete = DataSpecUtils::asConcreteDataMatcher(requested);
-    publisher.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec, requested.lifetime, requested.metadata);
-  }
+  requestedInputs |
+    views::filter_not_matching(providedOutputs) |   // filter the inputs that are already provided
+    views::filter_not_matching(publisher.outputs) | // filter the inputs that are already covered
+    views::input_to_output_specs() |
+    sinks::append_to{publisher.outputs}; // append them to the publisher outputs
 }
 
 void AnalysisSupportHelpers::addMissingOutputsToSpawner(std::vector<OutputSpec> const& providedSpecials,
                                                         std::vector<InputSpec> const& requestedSpecials,
                                                         std::vector<InputSpec>& requestedAODs,
                                                         DataProcessorSpec& publisher)
 {
-  for (auto& input : requestedSpecials) {
-    if (std::any_of(providedSpecials.begin(), providedSpecials.end(), [&input](auto const& x) {
-          return DataSpecUtils::match(input, x);
-        })) {
-      continue;
-    }
-    auto concrete = DataSpecUtils::asConcreteDataMatcher(input);
-    publisher.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec);
-    for (auto& i : input.metadata) {
-      if ((i.type == VariantType::String) && (i.name.find("input:") != std::string::npos)) {
-        auto spec = DataSpecUtils::fromMetadataString(i.defaultValue.get<std::string>());
-        auto j = std::find(publisher.inputs.begin(), publisher.inputs.end(), spec);
-        if (j == publisher.inputs.end()) {
-          publisher.inputs.push_back(spec);
-        }
-        DataSpecUtils::updateInputList(requestedAODs, std::move(spec));
-      }
-    }
+  requestedSpecials |
+    views::filter_not_matching(providedSpecials) | // filter the inputs that are already provided
+    views::input_to_output_specs() |
+    sinks::append_to{publisher.outputs}; // append them to the publisher outputs
+
+  std::vector<InputSpec> additionalInputs;
+  for (auto& input : requestedSpecials | views::filter_not_matching(providedSpecials)) {
+    input.metadata |
+      views::filter_string_params_with("input:") |
+      views::params_to_input_specs() |
+      sinks::update_input_list{additionalInputs}; // store into a temporary
   }
+  additionalInputs | sinks::update_input_list{requestedAODs};    // update requestedAODs
+  additionalInputs | sinks::update_input_list{publisher.inputs}; // update publisher inputs
 }
 
 void AnalysisSupportHelpers::addMissingOutputsToBuilder(std::vector<InputSpec> const& requestedSpecials,
                                                         std::vector<InputSpec>& requestedAODs,
                                                         std::vector<InputSpec>& requestedDYNs,
                                                         DataProcessorSpec& publisher)
 {
-  for (auto& input : requestedSpecials) {
-    auto concrete = DataSpecUtils::asConcreteDataMatcher(input);
-    publisher.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec);
-    for (auto& i : input.metadata) {
-      if ((i.type == VariantType::String) && (i.name.find("input:") != std::string::npos)) {
-        auto spec = DataSpecUtils::fromMetadataString(i.defaultValue.get<std::string>());
-        auto j = std::find_if(publisher.inputs.begin(), publisher.inputs.end(), [&](auto x) { return x.binding == spec.binding; });
-        if (j == publisher.inputs.end()) {
-          publisher.inputs.push_back(spec);
-        }
-        if (DataSpecUtils::partialMatch(spec, AODOrigins)) {
-          DataSpecUtils::updateInputList(requestedAODs, std::move(spec));
-        } else if (DataSpecUtils::partialMatch(spec, header::DataOrigin{"DYN"})) {
-          DataSpecUtils::updateInputList(requestedDYNs, std::move(spec));
-        }
-      }
-    }
+  requestedSpecials |
+    views::input_to_output_specs() |
+    sinks::append_to{publisher.outputs}; // append them to the publisher outputs
+
+  std::vector<InputSpec> additionalInputs;
+  for (auto const& input : requestedSpecials) {
+    input.metadata |
+      views::filter_string_params_with("input:") |
+      views::params_to_input_specs() |
+      sinks::update_input_list{additionalInputs}; // store into a temporary
   }
-}
 
-void AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher(
-  std::vector<OutputSpec> const& providedSpecials,
-  std::vector<InputSpec> const& requestedSpecials,
-  std::vector<InputSpec>& requestedAODs,
-  std::vector<InputSpec>& requestedDYNs,
-  DataProcessorSpec& publisher)
-{
-  for (auto& input : requestedSpecials) {
-    auto concrete = DataSpecUtils::asConcreteDataMatcher(input);
-    publisher.outputs.emplace_back(concrete.origin, concrete.description, concrete.subSpec);
-    // FIXME: good enough for now...
-    for (auto& i : input.metadata) {
-      if ((i.type == VariantType::String) && (i.name.find("input:") != std::string::npos)) {
-        auto spec = DataSpecUtils::fromMetadataString(i.defaultValue.get<std::string>());
-        auto j = std::find_if(publisher.inputs.begin(), publisher.inputs.end(), [&](auto x) { return x.binding == spec.binding; });
-        if (j == publisher.inputs.end()) {
-          publisher.inputs.push_back(spec);
-        }
-        if (DataSpecUtils::partialMatch(spec, AODOrigins)) {
-          DataSpecUtils::updateInputList(requestedAODs, std::move(spec));
-        } else if (DataSpecUtils::partialMatch(spec, header::DataOrigin{"DYN"})) {
-          DataSpecUtils::updateInputList(requestedDYNs, std::move(spec));
-        }
-      }
-    }
-  }
+  additionalInputs | sinks::update_input_list{publisher.inputs}; // update publisher inputs
+  // FIXME: until we have a single list of pairs
+  additionalInputs |
+    views::partial_match_filter(AODOrigins) |
+    sinks::update_input_list{requestedAODs}; // update requestedAODs
+  additionalInputs |
+    views::partial_match_filter(header::DataOrigin{"DYN"}) |
+    sinks::update_input_list{requestedDYNs}; // update requestedDYNs
 }
 
 // =============================================================================
diff --git a/Framework/Core/src/ArrowSupport.cxx b/Framework/Core/src/ArrowSupport.cxx
@@ -595,23 +595,16 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
       ac.providedTIMs.clear();
       ac.requestedTIMs.clear();
 
-
       auto inputSpecLessThan = [](InputSpec const& lhs, InputSpec const& rhs) { return DataSpecUtils::describe(lhs) < DataSpecUtils::describe(rhs); };
       auto outputSpecLessThan = [](OutputSpec const& lhs, OutputSpec const& rhs) { return DataSpecUtils::describe(lhs) < DataSpecUtils::describe(rhs); };
 
       if (builder != workflow.end()) {
         // collect currently requested IDXs
         ac.requestedIDXs.clear();
-        for (auto& d : workflow) {
-          if (d.name == builder->name) {
-            continue;
-          }
-          for (auto& i : d.inputs) {
-            if (DataSpecUtils::partialMatch(i, header::DataOrigin{"IDX"})) {
-              auto copy = i;
-              DataSpecUtils::updateInputList(ac.requestedIDXs, std::move(copy));
-            }
-          }
+        for (auto& d : workflow | views::exclude_by_name(builder->name)) {
+          d.inputs |
+            views::partial_match_filter(header::DataOrigin{"IDX"}) |
+            sinks::update_input_list{ac.requestedIDXs};
         }
         // recreate inputs and outputs
         builder->inputs.clear();
@@ -624,37 +617,27 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
 
       if (spawner != workflow.end()) {
         // collect currently requested DYNs
-        for (auto& d : workflow) {
-          if (d.name == spawner->name) {
-            continue;
-          }
-          for (auto const& i : d.inputs) {
-            if (DataSpecUtils::partialMatch(i, header::DataOrigin{"DYN"})) {
-              auto copy = i;
-              DataSpecUtils::updateInputList(ac.requestedDYNs, std::move(copy));
-            }
-          }
-          for (auto const& o : d.outputs) {
-            if (DataSpecUtils::partialMatch(o, header::DataOrigin{"DYN"})) {
-              ac.providedDYNs.emplace_back(o);
-            }
-          }
+        for (auto& d : workflow | views::exclude_by_name(spawner->name)) {
+          d.inputs |
+            views::partial_match_filter(header::DataOrigin{"DYN"}) |
+            sinks::update_input_list{ac.requestedDYNs};
+          d.outputs |
+            views::partial_match_filter(header::DataOrigin{"DYN"}) |
+            sinks::append_to{ac.providedDYNs};
         }
         std::sort(ac.requestedDYNs.begin(), ac.requestedDYNs.end(), inputSpecLessThan);
         std::sort(ac.providedDYNs.begin(), ac.providedDYNs.end(), outputSpecLessThan);
         ac.spawnerInputs.clear();
-        for (auto& input : ac.requestedDYNs) {
-          if (std::none_of(ac.providedDYNs.begin(), ac.providedDYNs.end(), [&input](auto const& x) { return DataSpecUtils::match(input, x); })) {
-            ac.spawnerInputs.emplace_back(input);
-          }
-        }
+        ac.requestedDYNs |
+          views::filter_not_matching(ac.providedDYNs) |
+          sinks::append_to{ac.spawnerInputs};
         // recreate inputs and outputs
         spawner->outputs.clear();
         spawner->inputs.clear();
+        AnalysisSupportHelpers::addMissingOutputsToSpawner({}, ac.spawnerInputs, ac.requestedAODs, *spawner);
         // replace AlgorithmSpec
         // FIXME: it should be made more generic, so it does not need replacement...
         spawner->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "ExtendedTableSpawner", ctx);
-        AnalysisSupportHelpers::addMissingOutputsToSpawner({}, ac.spawnerInputs, ac.requestedAODs, *spawner);
       }
 
       if (analysisCCDB != workflow.end()) {
@@ -675,7 +658,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         // FIXME: it should be made more generic, so it does not need replacement...
         // FIXME how can I make the lookup depend on DYN tables as well??
         analysisCCDB->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "AnalysisCCDBFetcherPlugin", ctx);
-        AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher({}, ac.analysisCCDBInputs, ac.requestedAODs, ac.requestedDYNs, *analysisCCDB);
+        AnalysisSupportHelpers::addMissingOutputsToBuilder(ac.analysisCCDBInputs, ac.requestedAODs, ac.requestedDYNs, *analysisCCDB);
       }
 
       if (writer != workflow.end()) {
@@ -686,12 +669,9 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         // If reader and/or builder were adjusted, remove unneeded outputs
         // update currently requested AODs
         for (auto& d : workflow) {
-          for (auto const& i : d.inputs) {
-            if (DataSpecUtils::partialMatch(i, AODOrigins)) {
-              auto copy = i;
-              DataSpecUtils::updateInputList(ac.requestedAODs, std::move(copy));
-            }
-          }
+          d.inputs |
+            views::partial_match_filter(AODOrigins) |
+            sinks::update_input_list{ac.requestedAODs};
         }
 
         // remove unmatched outputs
@@ -705,8 +685,6 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         }
       }
 
-
-
       // replace writer as some outputs may have become dangling and some are now consumed
       auto [outputsInputs, isDangling] = WorkflowHelpers::analyzeOutputs(workflow);
 
diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx
