From eb2c03e28bc92d924eb8f6138cdc0cddaa6ba57b Mon Sep 17 00:00:00 2001 From: Anton Alkin Date: Fri, 16 Oct 2026 13:05:29 +0200 Subject: [PATCH 1/2] 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. --- Framework/Core/src/ArrowSupport.cxx | 29 +----------------------- Framework/Core/src/runDataProcessing.cxx | 29 ++++++++++++++++++++++-- 2 files changed, 28 insertions(+), 30 deletions(-) diff --git a/Framework/Core/src/ArrowSupport.cxx b/Framework/Core/src/ArrowSupport.cxx index cf72696ff7445..584f7397e67f8 100644 --- a/Framework/Core/src/ArrowSupport.cxx +++ b/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 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); } diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx index fb4d8593e528a..40aba8234aa4c 100644 --- a/Framework/Core/src/runDataProcessing.cxx +++ b/Framework/Core/src/runDataProcessing.cxx @@ -1673,7 +1673,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()) { @@ -1684,6 +1684,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 ccdbPaths; for (auto& device : altered_workflow) { // ignore internal devices if (device.name.find("internal") != std::string::npos) { @@ -1718,14 +1720,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(confName.c_str()); } + } else if (param.type == VariantType::String && param.name.starts_with("ccdb:")) { + // update inputs ccdb urls from configuration + param.defaultValue = reg->get(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()) { + throw runtime_error_f("Task {} has a CCDB path inconsistent with another found in same workflow: {} vs {}", + device.name.c_str(), + param.defaultValue.get().c_str(), + ccdbPaths[param.name].c_str()); + } + } else { + ccdbPaths.emplace(param.name, param.defaultValue.get()); + } + } + } + /// FIXME: use commandline arguments as alternative LOGP(debug, "Original inputs: "); for (auto& input : device.inputs) { From 3629cfc114e20ebb5707589f77fbb520e235cf92 Mon Sep 17 00:00:00 2001 From: Anton Alkin Date: Fri, 16 Oct 2026 14:12:31 +0200 Subject: [PATCH 2/2] fix condition and error message --- Framework/Core/src/runDataProcessing.cxx | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx index 40aba8234aa4c..3f746b1d76f48 100644 --- a/Framework/Core/src/runDataProcessing.cxx +++ b/Framework/Core/src/runDataProcessing.cxx @@ -1735,15 +1735,16 @@ int runStateMachine(DataProcessorSpecs const& workflow, // 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:")) { + if (param.type != VariantType::String || !param.name.starts_with("ccdb:")) { continue; } if (ccdbPaths.contains(param.name)) { if (ccdbPaths[param.name] != param.defaultValue.get()) { - throw runtime_error_f("Task {} has a CCDB path inconsistent with another found in same workflow: {} vs {}", - device.name.c_str(), - param.defaultValue.get().c_str(), - ccdbPaths[param.name].c_str()); + throw runtime_error_f(fmt::format("Task {} has a CCDB path inconsistent with another found in same workflow: {} vs {}", + device.name.c_str(), + param.defaultValue.get().c_str(), + ccdbPaths[param.name].c_str()) + .c_str()); } } else { ccdbPaths.emplace(param.name, param.defaultValue.get());