DPL: allow for configurable CCDB paths (#15337)

Author: ktf

Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx

@@ -21,7 +21,7 @@
 #include "Framework/Signpost.h"
 #include "Framework/DanglingEdgesContext.h"
 #include "Framework/ConfigContext.h"
-#include "Framework/ConfigContext.h"
+#include "Framework/ConfigParamsHelper.h"
 #include <arrow/array/builder_binary.h>
 #include <arrow/type.h>
 #include <arrow/type_fwd.h>
@@ -71,31 +71,45 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
 {
   return adaptStateful([](ConfigParamRegistry const& options, DeviceSpec const& spec, InitContext& ic) {
     auto& dec = ic.services().get<DanglingEdgesContext>();
+    // The effective default for each ccdb: option was already resolved at topology
+    // time by ArrowSupport (consulting task Configurables) and registered on this
+    // device's options. Here we just read the final value — honouring any further
+    // runtime override supplied via CLI or JSON config.
+    std::unordered_map<std::string, std::string> ccdbUrls;
+    for (auto& input : dec.analysisCCDBInputs) {
+      for (auto& m : input.metadata) {
+        if (!m.name.starts_with("ccdb:") || ccdbUrls.count(m.name)) {
+          continue;
+        }
+        std::string url = m.defaultValue.asString();
+        if (ConfigParamsHelper::hasOption(spec.options, m.name)) {
+          url = options.get<std::string>(m.name.c_str());
+        }
+        LOGP(info, "CCDB path resolved for {}: {}", m.name, url);
+        ccdbUrls.emplace(m.name, std::move(url));
+      }
+    }
     std::vector<std::shared_ptr<arrow::Schema>> schemas;
-    auto schemaMetadata = std::make_shared<arrow::KeyValueMetadata>();
-
     for (auto& input : dec.analysisCCDBInputs) {
+      auto schemaMetadata = std::make_shared<arrow::KeyValueMetadata>();
       std::vector<std::shared_ptr<arrow::Field>> fields;
       schemaMetadata->Append("outputRoute", DataSpecUtils::describe(input));
       schemaMetadata->Append("outputBinding", input.binding);
-
       for (auto& m : input.metadata) {
-        // Save the list of input tables
         if (m.name.starts_with("input:")) {
           auto name = m.name.substr(6);
           schemaMetadata->Append("sourceTable", name);
           schemaMetadata->Append("sourceMatcher", DataSpecUtils::describe(std::get<ConcreteDataMatcher>(DataSpecUtils::fromMetadataString(m.defaultValue.get<std::string>()).matcher)));
           continue;
         }
-        // Ignore the non ccdb: entries
         if (!m.name.starts_with("ccdb:")) {
           continue;
         }
-        // Create the schema of the output
-        auto metadata = std::make_shared<arrow::KeyValueMetadata>();
-        metadata->Append("url", m.defaultValue.asString());
+        auto fieldMetadata = std::make_shared<arrow::KeyValueMetadata>();
+        auto it = ccdbUrls.find(m.name);
+        fieldMetadata->Append("url", it != ccdbUrls.end() ? it->second : m.defaultValue.asString());
         auto columnName = m.name.substr(strlen("ccdb:"));
-        fields.emplace_back(std::make_shared<arrow::Field>(columnName, arrow::binary_view(), false, metadata));
+        fields.emplace_back(std::make_shared<arrow::Field>(columnName, arrow::binary_view(), false, fieldMetadata));
       }
       schemas.emplace_back(std::make_shared<arrow::Schema>(fields, schemaMetadata));
     }

Framework/Core/include/Framework/AnalysisHelpers.h

@@ -172,7 +172,7 @@ struct Builder {
 
   std::shared_ptr<arrow::Table> materialize(ProcessingContext& pc);
 };
-}  // namespace o2::framework
+} // namespace o2::framework
 
 namespace o2::soa
 {
@@ -394,7 +394,7 @@ constexpr auto getIndexMetadata() -> std::vector<framework::ConfigParamSpec>
   return {};
 }
 
-}  // namespace
+} // namespace
 
 template <TableRef R>
 constexpr auto tableRef2InputSpec()
@@ -463,7 +463,7 @@ constexpr auto tableRef2OutputRef()
     o2::aod::label<R>(),
     R.version};
 }
-}  // namespace o2::soa
+} // namespace o2::soa
 
 namespace o2::framework
 {
@@ -672,7 +672,7 @@ struct Spawns : decltype(transformBase<T>()) {
 
   std::shared_ptr<typename T::table_t> table = nullptr;
   std::shared_ptr<extension_t> extension = nullptr;
-  std::array<o2::framework::expressions::Projector, N> projectors = []<typename... C>(framework::pack<C...>) -> std::array<expressions::Projector, sizeof...(C)>
+  std::array<o2::framework::expressions::Projector, N> projectors = []<typename... C>(framework::pack<C...>)->std::array<expressions::Projector, sizeof...(C)>
   {
     return {{std::move(C::Projector())...}};
   }
@@ -1077,7 +1077,7 @@ concept is_partition = requires(T t) {
   requires std::same_as<decltype(t.filter), expressions::Filter>;
   requires std::same_as<decltype(t.mFiltered), std::unique_ptr<o2::soa::Filtered<typename T::content_t>>>;
 };
-}  // namespace o2::framework
+} // namespace o2::framework
 
 namespace o2::soa
 {
@@ -1100,6 +1100,6 @@ auto Attach(T const& table)
   using output_t = Join<T, o2::soa::Table<o2::aod::Hash<"JOIN"_h>, o2::aod::Hash<"JOIN/0"_h>, o2::aod::Hash<"JOIN"_h>, Cs...>>;
   return output_t{{table.asArrowTable()}, table.offset()};
 }
-}  // namespace o2::soa
+} // namespace o2::soa
 
-#endif  // o2_framework_AnalysisHelpers_H_DEFINED
+#endif // o2_framework_AnalysisHelpers_H_DEFINED

Framework/Core/include/Framework/Configurable.h

@@ -83,6 +83,26 @@ struct Configurable : IP {
 template <typename T, ConfigParamKind K = ConfigParamKind::kGeneric>
 using MutableConfigurable = Configurable<T, K, ConfigurablePolicyMutable<T, K>>;
 
+/// Convenience wrapper for overriding the CCDB path of a CCDB column declared
+/// with DECLARE_SOA_CCDB_COLUMN / DECLARE_SOA_CCDB_COLUMN_FULL.
+///
+/// The option name, default value, and help string are all derived automatically
+/// from the column type: name = "ccdb:" + Column::mLabel, default = Column::query.
+///
+/// Example:
+///   struct MyTask {
+///     ConfigurableCCDBPath<tofcalib::LHCphase> lhcPhasePath;
+///   };
+template <typename Column>
+struct ConfigurableCCDBPath : Configurable<std::string> {
+  ConfigurableCCDBPath()
+    : Configurable<std::string>{std::string{"ccdb:"} + Column::mLabel,
+                                std::string{Column::query},
+                                std::string{"CCDB path for "} + Column::mLabel + " (default: " + Column::query + ")"}
+  {
+  }
+};
+
 template <typename T>
 concept is_configurable = requires(T t) {
   requires std::same_as<std::string, decltype(t.name)>;
@@ -93,11 +113,10 @@ concept is_configurable = requires(T t) {
 using ConfigurableAxis = Configurable<std::vector<double>, ConfigParamKind::kAxisSpec, ConfigurablePolicyConst<std::vector<double>, ConfigParamKind::kAxisSpec>>;
 
 template <typename T>
-concept is_configurable_axis = is_configurable<T>&&
-  requires()
-{
-  T::kind == ConfigParamKind::kAxisSpec;
-};
+concept is_configurable_axis = is_configurable<T> &&
+                               requires() {
+                                 T::kind == ConfigParamKind::kAxisSpec;
+                               };
 
 template <typename T, typename... As>
 struct ProcessConfigurable : Configurable<bool, ConfigParamKind::kProcessFlag> {

Framework/Core/src/AnalysisHelpers.cxx

@@ -201,4 +201,5 @@ std::shared_ptr<arrow::Table> Builder::materialize(ProcessingContext& pc)
   result = o2::soa::IndexBuilder::materialize(*builders.get(), std::move(tables), records, outputSchema, exclusive);
   return result;
 }
+
 } // namespace o2::framework

Framework/Core/src/ArrowSupport.cxx

@@ -34,6 +34,7 @@
 #include "Framework/ServiceRegistryHelpers.h"
 #include "Framework/Signpost.h"
 #include "Framework/DefaultsHelpers.h"
+#include "Framework/ConfigParamsHelper.h"
 
 #include "CommonMessageBackendsHelpers.h"
 #include <Monitoring/Monitoring.h>
@@ -637,6 +638,34 @@ 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/TestWorkflows/src/o2TestAnalysisCCDB.cxx

@@ -51,19 +51,35 @@ struct DummyTimestampsTable {
 };
 
 struct SimpleCCDBConsumer {
+  ConfigurableCCDBPath<o2::aod::tofcalib::LHCphase> lhcPhasePath;
+
   void process(o2::aod::TOFCalibrationObjects const& ccdbObjectsForAllTimestamps)
   {
+    LOGP(info, "LHCphase CCDB path configurable value: {}", lhcPhasePath.value);
     LOGP(info, "Looking at all the LHCphases associated to the timestamps");
     for (auto& object : ccdbObjectsForAllTimestamps) {
       std::cout << object.lhcPhase().getStartValidity() << " " << object.lhcPhase().getEndValidity() << std::endl;
     }
   }
 };
 
+struct AnotherCCDBConsumer {
+  ConfigurableCCDBPath<o2::aod::tofcalib::LHCphase> lhcPhasePath;
+
+  void process(o2::aod::TOFCalibrationObjects const& ccdbObjectsForAllTimestamps)
+  {
+    LOGP(info, "AnotherCCDBConsumer LHCphase CCDB path configurable value: {}", lhcPhasePath.value);
+    for (auto& object : ccdbObjectsForAllTimestamps) {
+      std::cout << object.lhcPhase().getStartValidity() << " " << object.lhcPhase().getEndValidity() << std::endl;
+    }
+  }
+};
+
 WorkflowSpec defineDataProcessing(ConfigContext const& cfgc)
 {
   return WorkflowSpec{
     adaptAnalysisTask<DummyTimestampsTable>(cfgc),
-    adaptAnalysisTask<SimpleCCDBConsumer>(cfgc, TaskName{"simple-ccdb-cunsumer"}),
+    adaptAnalysisTask<SimpleCCDBConsumer>(cfgc, TaskName{"simple-ccdb-consumer"}),
+    adaptAnalysisTask<AnotherCCDBConsumer>(cfgc, TaskName{"another-ccdb-consumer"}),
   };
 }