DPL Analysis: update the ccdb path options from configuration

During MATERIALIZE_WORKFLOW the read configuration is used to update
ccdb urls in in tasks' options, so that later analysis ccdb fetcher will
get updated paths in adjustTopology.

Author: aalkin

Framework/Core/src/ArrowSupport.cxx

@@ -638,34 +638,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         analysisCCDB->outputs.clear();
         analysisCCDB->inputs.clear();
         AnalysisSupportHelpers::addMissingOutputsToBuilder(dec.analysisCCDBInputs, dec.requestedAODs, dec.requestedDYNs, *analysisCCDB);
-        // Register each ccdb: column path as an actual device option on the CCDB
-        // device so it can be read from ConfigParamRegistry at runtime.
-        // If any analysis task declared a Configurable<std::string> with the same
-        // "ccdb:fXxx" name, prefer its default over the compile-time ::query value.
-        // First encountered wins; log a warning if two tasks declare conflicting defaults.
-        for (auto& input : dec.analysisCCDBInputs) {
-          for (auto& m : input.metadata | std::views::filter(checks::has_params_with_name_starting("ccdb:"))) {
-            ConfigParamSpec effective = m; // start with compile-time default
-            bool foundFirst = false;
-            for (auto& d : workflow | views::exclude_by_name(analysisCCDB->name)) {
-              for (auto& opt : d.options) {
-                if (opt.name == m.name) {
-                  if (!foundFirst) {
-                    effective = opt; // first task Configurable wins
-                    foundFirst = true;
-                  } else if (opt.defaultValue.asString() != effective.defaultValue.asString()) {
-                    LOGP(warn, "Task '{}' declares Configurable '{}' = '{}' which conflicts "
-                               "with an earlier value '{}'; earlier value will be used.",
-                         d.name, opt.name, opt.defaultValue.asString(),
-                         effective.defaultValue.asString());
-                  }
-                  break;
-                }
-              }
-            }
-            ConfigParamsHelper::addOptionIfMissing(analysisCCDB->options, effective);
-          }
-        }
+
         // load real AlgorithmSpec before deployment
         analysisCCDB->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "AnalysisCCDBFetcherPlugin", ctx);
       }

Framework/Core/src/runDataProcessing.cxx

@@ -1672,7 +1672,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
           /// prune device inputs
           auto altered_workflow = workflow;
 
-          auto confNameFromParam = [](std::string const& paramName) {
+          auto controlParamFromName = [](std::string const& paramName) {
             std::regex name_regex(R"(^control:([\w-]+)\/(\w+))");
             auto match = std::sregex_token_iterator(paramName.begin(), paramName.end(), name_regex, 0);
             if (match == std::sregex_token_iterator()) {
@@ -1683,6 +1683,8 @@ int runStateMachine(DataProcessorSpecs const& workflow,
             return std::pair{task, conf};
           };
           bool altered = false;
+          // verify uniformity of ccdb request for a each unique column
+          std::unordered_map<std::string, std::string> ccdbPaths;
           for (auto& device : altered_workflow) {
             // ignore internal devices
             if (device.name.find("internal") != std::string::npos) {
@@ -1717,14 +1719,37 @@ int runStateMachine(DataProcessorSpecs const& workflow,
               for (auto& input : device.inputs) {
                 for (auto& param : input.metadata) {
                   if (param.type == VariantType::Bool && param.name.starts_with("control:")) {
+                    // update inputs from process switches
                     if (param.name != "control:default" && param.name != "control:spawn" && param.name != "control:build" && param.name != "control:define") {
-                      auto confName = confNameFromParam(param.name).second;
+                      auto confName = controlParamFromName(param.name).second;
                       param.defaultValue = reg->get<bool>(confName.c_str());
                     }
+                  } else if (param.type == VariantType::String && param.name.starts_with("ccdb:")) {
+                    // update inputs ccdb urls from configuration
+                    param.defaultValue = reg->get<std::string>(param.name.c_str());
                   }
                 }
               }
             }
+            // after updating the urls, verify the consistency
+            for (auto& input : device.inputs) {
+              for (auto& param : input.metadata) {
+                if (param.type != VariantType::String || param.name.starts_with("ccdb:")) {
+                  continue;
+                }
+                if (ccdbPaths.contains(param.name)) {
+                  if (ccdbPaths[param.name] != param.defaultValue.get<std::string>()) {
+                    throw runtime_error_f("Task {} has a CCDB path inconsistent with another found in same workflow: {} vs {}",
+                                          device.name.c_str(),
+                                          param.defaultValue.get<std::string>().c_str(),
+                                          ccdbPaths[param.name].c_str());
+                  }
+                } else {
+                  ccdbPaths.emplace(param.name, param.defaultValue.get<std::string>());
+                }
+              }
+            }
+
             /// FIXME: use commandline arguments as alternative
             LOGP(debug, "Original inputs: ");
             for (auto& input : device.inputs) {