Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 1 addition & 28 deletions Framework/Core/src/ArrowSupport.cxx
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
30 changes: 28 additions & 2 deletions Framework/Core/src/runDataProcessing.cxx
Original file line number Diff line number Diff line change
Expand Up @@ -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()) {
Expand All @@ -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<std::string, std::string> ccdbPaths;
for (auto& device : altered_workflow) {
// ignore internal devices
if (device.name.find("internal") != std::string::npos) {
Expand Down Expand Up @@ -1718,14 +1720,38 @@ 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(fmt::format("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())
.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) {
Expand Down
Loading