WIP DPL Analysis: centralised CCDB support in analysis

ktf · ktf · commit 35a19078471a · 2025-08-16T23:38:06.000+02:00
Thanks to the newly added binary view columns we can finally support
proper CCDB integration in analysis.

In order to do so, the user needs to create a TIMESTAMPED table, i.e. a table
which is an extension of another one where the timestamps for each rows
are provided. The extra columns of such timestamped table will be CCDB columns
where the iterator of each provides access for one specified CCDB object.
diff --git a/Framework/CCDBSupport/CMakeLists.txt b/Framework/CCDBSupport/CMakeLists.txt
@@ -1,4 +1,4 @@
-# Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+# Copyright 2019-2025 CERN and copyright holders of ALICE O2.
 # See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
 # All rights not expressly granted are reserved.
 #
@@ -9,9 +9,10 @@
 # granted to it by virtue of its status as an Intergovernmental Organization
 # or submit itself to any jurisdiction.
 o2_add_library(FrameworkCCDBSupport
-               SOURCES 
+               SOURCES
                 src/Plugin.cxx
                 src/CCDBHelpers.cxx
+                src/AnalysisCCDBHelpers.cxx
                PRIVATE_INCLUDE_DIRECTORIES ${CMAKE_CURRENT_LIST_DIR}/src
                PUBLIC_LINK_LIBRARIES O2::Framework O2::CCDB)
 
diff --git a/Framework/CCDBSupport/src/Plugin.cxx b/Framework/CCDBSupport/src/Plugin.cxx
@@ -10,6 +10,7 @@
 // or submit itself to any jurisdiction.
 #include "Framework/Plugins.h"
 #include "Framework/AlgorithmSpec.h"
+#include "AnalysisCCDBHelpers.h"
 #include "CCDBHelpers.h"
 
 struct CCDBFetcherPlugin : o2::framework::AlgorithmPlugin {
@@ -19,6 +20,14 @@ struct CCDBFetcherPlugin : o2::framework::AlgorithmPlugin {
   }
 };
 
+struct AnalysisCCDBFetcherPlugin : o2::framework::AlgorithmPlugin {
+  o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& ctx) final
+  {
+    return o2::framework::AnalysisCCDBHelpers::fetchFromCCDB(ctx);
+  }
+};
+
 DEFINE_DPL_PLUGINS_BEGIN
 DEFINE_DPL_PLUGIN_INSTANCE(CCDBFetcherPlugin, CustomAlgorithm);
+DEFINE_DPL_PLUGIN_INSTANCE(AnalysisCCDBFetcherPlugin, CustomAlgorithm);
 DEFINE_DPL_PLUGINS_END
diff --git a/Framework/Core/include/Framework/ASoA.h b/Framework/Core/include/Framework/ASoA.h
@@ -1274,6 +1274,11 @@ concept with_sources = requires {
   T::sources.size();
 };
 
+template <typename T>
+concept with_ccdb_urls = requires {
+  T::ccdb_urls.size();
+};
+
 template <typename T>
 concept with_base_table = not_void<typename aod::MetadataTrait<o2::aod::Hash<T::ref.desc_hash>>::metadata::base_table_t>;
 
@@ -2248,11 +2253,14 @@ ColumnGetterFunction<R, typename T::iterator> getColumnGetterByLabel(const std::
 
 namespace o2::aod
 {
+// If you get an error about not satisfying is_origin_hash, you need to add
+// an entry here.
 O2ORIGIN("AOD");
 O2ORIGIN("AOD1");
 O2ORIGIN("AOD2");
 O2ORIGIN("DYN");
 O2ORIGIN("IDX");
+O2ORIGIN("TIM");
 O2ORIGIN("JOIN");
 O2HASH("JOIN/0");
 O2ORIGIN("CONC");
@@ -2313,6 +2321,46 @@ consteval static std::string_view namespace_prefix()
   };                                                                                                                                                                              \
   [[maybe_unused]] static constexpr o2::framework::expressions::BindingNode _Getter_ { _Label_, _Name_::hash, o2::framework::expressions::selectArrowType<_Type_>() }
 
+#define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_)                 \
+  struct _Name_ : o2::soa::Column<std::span<std::byte>, _Name_> {                                            \
+    static constexpr const char* mLabel = _Label_;                                                           \
+    static constexpr const char* query = _CCDBQuery_;                                                        \
+    static constexpr const uint32_t hash = crc32(namespace_prefix<_Name_>(), std::string_view{#_Getter_});   \
+    using base = o2::soa::Column<std::span<std::byte>, _Name_>;                                              \
+    using type = std::span<std::byte>;                                                                       \
+    using column_t = _Name_;                                                                                 \
+    _Name_(arrow::ChunkedArray const* column)                                                                \
+      : o2::soa::Column<std::span<std::byte>, _Name_>(o2::soa::ColumnIterator<std::span<std::byte>>(column)) \
+    {                                                                                                        \
+    }                                                                                                        \
+                                                                                                             \
+    _Name_() = default;                                                                                      \
+    _Name_(_Name_ const& other) = default;                                                                   \
+    _Name_& operator=(_Name_ const& other) = default;                                                        \
+                                                                                                             \
+    decltype(auto) _Getter_() const                                                                          \
+    {                                                                                                        \
+      static std::byte* payload = nullptr;                                                                   \
+      static _ConcreteType_* deserialised = nullptr;                                                         \
+      auto span = *mColumnIterator;                                                                          \
+      if (payload != (std::byte*)span.data()) {                                                              \
+        payload = (std::byte*)span.data();                                                                   \
+        TBufferFile f(TBufferFile::EMode::kRead, span.size(), (char*)span.data(), kFALSE);                   \
+        deserialised = (_ConcreteType_*)f.ReadObjectAny(TClass::GetClass(#_ConcreteType_));                  \
+      }                                                                                                      \
+      return *deserialised;                                                                                  \
+    }                                                                                                        \
+                                                                                                             \
+    decltype(auto)                                                                                           \
+      get() const                                                                                            \
+    {                                                                                                        \
+      return _Getter_();                                                                                     \
+    }                                                                                                        \
+  };
+
+#define DECLARE_SOA_CCDB_COLUMN(_Name_, _Getter_, _ConcreteType_, _CCDBQuery_) \
+  DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, "f" #_Name_, _Getter_, _ConcreteType_, _CCDBQuery_)
+
 #define DECLARE_SOA_COLUMN(_Name_, _Getter_, _Type_) \
   DECLARE_SOA_COLUMN_FULL(_Name_, _Getter_, _Type_, "f" #_Name_)
 
@@ -3188,6 +3236,43 @@ consteval auto getIndexTargets()
     using metadata = _Name_##Metadata;                                                                                                     \
   };
 
+// Declare were each row is associated to a timestamp column of an _TimestampSource_
+// table.
+//
+// The columns of this table have to be CCDB_COLUMNS so that for each timestamp, we get a row
+// which points to the specified CCDB objectes described by those columns.
+#define DECLARE_SOA_TIMESTAMPED_TABLE_FULL(_Name_, _Label_, _TimestampSource_, _TimestampColumn_, _Origin_, _Version_, _Desc_, ...) \
+  O2HASH(_Desc_ "/" #_Version_);                                                                                                    \
+  template <typename O>                                                                                                             \
+  using _Name_##TimestampFrom = soa::Table<o2::aod::Hash<_Label_ ""_h>, o2::aod::Hash<_Desc_ "/" #_Version_ ""_h>, O>;              \
+  using _Name_##Timestamp = _Name_##TimestampFrom<o2::aod::Hash<_Origin_ ""_h>>;                                                    \
+  template <typename O = o2::aod::Hash<_Origin_ ""_h>>                                                                              \
+  struct _Name_##TimestampMetadataFrom : TableMetadata<o2::aod::Hash<_Desc_ "/" #_Version_ ""_h>, __VA_ARGS__> {                    \
+    using base_table_t = _TimestampSource_;                                                                                         \
+    using extension_table_t = _Name_##TimestampFrom<O>;                                                                             \
+    static constexpr const auto ccdb_urls = []<typename... Cs>(framework::pack<Cs...>) {                                            \
+      return std::array<std::string_view, sizeof...(Cs)>{Cs::query...};                                                             \
+    }(framework::pack<__VA_ARGS__>{});                                                                                              \
+    static constexpr const auto ccdb_bindings = []<typename... Cs>(framework::pack<Cs...>) {                                        \
+      return std::array<std::string_view, sizeof...(Cs)>{Cs::mLabel...};                                                            \
+    }(framework::pack<__VA_ARGS__>{});                                                                                              \
+    static constexpr auto sources = _TimestampSource_::originals;                                                                   \
+    static constexpr auto timestamp_column_label = _TimestampColumn_::mLabel;                                                       \
+    /*static constexpr auto timestampColumn = _TimestampColumn_;*/                                                                  \
+  };                                                                                                                                \
+  using _Name_##TimestampMetadata = _Name_##TimestampMetadataFrom<o2::aod::Hash<_Origin_ ""_h>>;                                    \
+  template <>                                                                                                                       \
+  struct MetadataTrait<o2::aod::Hash<_Desc_ "/" #_Version_ ""_h>> {                                                                 \
+    using metadata = _Name_##TimestampMetadata;                                                                                     \
+  };                                                                                                                                \
+  template <typename O>                                                                                                             \
+  using _Name_##From = o2::soa::JoinFull<o2::aod::Hash<_Desc_ "/" #_Version_ ""_h>, _TimestampSource_, _Name_##TimestampFrom<O>>;   \
+  using _Name_ = _Name_##From<o2::aod::Hash<_Origin_ ""_h>>;
+
+#define DECLARE_SOA_TIMESTAMPED_TABLE(_Name_, _TimestampSource_, _TimestampColumn_, _Version_, _Desc_, ...) \
+  O2HASH(#_Name_ "Timestamped");                                                                            \
+  DECLARE_SOA_TIMESTAMPED_TABLE_FULL(_Name_, #_Name_ "Timestamped", _TimestampSource_, _TimestampColumn_, "TIM", _Version_, _Desc_, __VA_ARGS__)
+
 #define DECLARE_SOA_INDEX_TABLE(_Name_, _Key_, _Description_, ...) \
   DECLARE_SOA_INDEX_TABLE_FULL(_Name_, _Key_, "IDX", 0, _Description_, false, __VA_ARGS__)
 
diff --git a/Framework/Core/include/Framework/AnalysisContext.h b/Framework/Core/include/Framework/AnalysisContext.h
@@ -29,16 +29,24 @@ struct OutputObjectInfo {
   std::vector<std::string> bindings;
 };
 
-//
+// This will keep track of the inputs which have
+// been requested and for which we will need to inject
+// some source device.
 struct AnalysisContext {
   std::vector<InputSpec> requestedAODs;
   std::vector<OutputSpec> providedAODs;
   std::vector<InputSpec> requestedDYNs;
   std::vector<OutputSpec> providedDYNs;
   std::vector<InputSpec> requestedIDXs;
+  std::vector<OutputSpec> providedTIMs;
+  std::vector<InputSpec> requestedTIMs;
   std::vector<OutputSpec> providedOutputObjHist;
   std::vector<InputSpec> spawnerInputs;
 
+  // These are the timestamped tables which are required to
+  // inject the the CCDB objecs.
+  std::vector<InputSpec> analysisCCDBInputs;
+
   // Needed to created the hist writer
   std::vector<OutputTaskInfo> outTskMap;
   std::vector<OutputObjectInfo> outObjHistMap;
diff --git a/Framework/Core/include/Framework/AnalysisHelpers.h b/Framework/Core/include/Framework/AnalysisHelpers.h
@@ -11,6 +11,7 @@
 #ifndef o2_framework_AnalysisHelpers_H_DEFINED
 #define o2_framework_AnalysisHelpers_H_DEFINED
 
+#include "ConfigParamSpec.h"
 #include "Framework/ASoA.h"
 #include "Framework/DataAllocator.h"
 #include "Framework/IndexBuilderHelpers.h"
@@ -49,6 +50,19 @@ inline constexpr auto getSources()
   }.template operator()<T::sources.size(), T::sources>();
 }
 
+template <soa::with_ccdb_urls T>
+inline constexpr auto getCCDBUrls()
+{
+  std::vector<framework::ConfigParamSpec> result;
+  for (size_t i = 0; i < T::ccdb_urls.size(); ++i) {
+    result.push_back({std::string{"ccdb:"} + std::string{T::ccdb_bindings[i]},
+                      framework::VariantType::String,
+                      T::ccdb_urls[i],
+                      {"\"\""}});
+  }
+  return result;
+}
+
 template <soa::with_sources T>
 constexpr auto getInputMetadata() -> std::vector<framework::ConfigParamSpec>
 {
@@ -67,18 +81,40 @@ constexpr auto getInputMetadata() -> std::vector<framework::ConfigParamSpec>
 {
   return {};
 }
+
+template <soa::with_ccdb_urls T>
+constexpr auto getCCDBMetadata() -> std::vector<framework::ConfigParamSpec>
+{
+  std::vector<framework::ConfigParamSpec> results = getCCDBUrls<T>();
+  std::sort(results.begin(), results.end(), [](framework::ConfigParamSpec const& a, framework::ConfigParamSpec const& b) { return a.name < b.name; });
+  auto last = std::unique(results.begin(), results.end(), [](framework::ConfigParamSpec const& a, framework::ConfigParamSpec const& b) { return a.name == b.name; });
+  results.erase(last, results.end());
+  return results;
+}
+
+template <typename T>
+constexpr auto getCCDBMetadata() -> std::vector<framework::ConfigParamSpec>
+{
+  return {};
+}
 }  // namespace
 
 template <TableRef R>
 constexpr auto tableRef2InputSpec()
 {
+  std::vector<framework::ConfigParamSpec> metadata;
+  auto m = getInputMetadata<typename o2::aod::MetadataTrait<o2::aod::Hash<R.desc_hash>>::metadata>();
+  metadata.insert(metadata.end(), m.begin(), m.end());
+  auto ccdbMetadata = getCCDBMetadata<typename o2::aod::MetadataTrait<o2::aod::Hash<R.desc_hash>>::metadata>();
+  metadata.insert(metadata.end(), ccdbMetadata.begin(), ccdbMetadata.end());
+
   return framework::InputSpec{
     o2::aod::label<R>(),
     o2::aod::origin<R>(),
     o2::aod::description(o2::aod::signature<R>()),
     R.version,
     framework::Lifetime::Timeframe,
-    getInputMetadata<typename o2::aod::MetadataTrait<o2::aod::Hash<R.desc_hash>>::metadata>()};
+    metadata};
 }
 
 template <TableRef R>
diff --git a/Framework/Core/include/Framework/AnalysisSupportHelpers.h b/Framework/Core/include/Framework/AnalysisSupportHelpers.h
@@ -39,6 +39,11 @@ 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/AnalysisTask.h b/Framework/Core/include/Framework/AnalysisTask.h
@@ -65,7 +65,8 @@ concept is_enumeration = is_enumeration_v<std::decay_t<T>>;
 
 // Helper struct which builds a DataProcessorSpec from
 // the contents of an AnalysisTask...
-namespace {
+namespace
+{
 struct AnalysisDataProcessorBuilder {
   template <soa::is_iterator G, typename... Args>
   static void addGroupingCandidates(Cache& bk, Cache& bku, bool enabled)
@@ -417,7 +418,7 @@ struct AnalysisDataProcessorBuilder {
     std::invoke(processingFunction, task, g, std::get<A>(at)...);
   }
 };
-}
+} // namespace
 
 struct SetDefaultProcesses {
   std::vector<std::pair<std::string, bool>> map;
@@ -429,7 +430,8 @@ struct TaskName {
   std::string value;
 };
 
-namespace {
+namespace
+{
 template <typename T, typename... A>
 auto getTaskNameSetProcesses(std::string& outputName, TaskName first, SetDefaultProcesses second, A... args)
 {
@@ -493,7 +495,7 @@ auto getTaskNameSetProcesses(std::string& outputName, A... args)
   return task;
 }
 
-}
+} // namespace
 
 /// Adaptor to make an AlgorithmSpec from a o2::framework::Task
 ///
diff --git a/Framework/Core/src/AnalysisSupportHelpers.cxx b/Framework/Core/src/AnalysisSupportHelpers.cxx
@@ -207,6 +207,35 @@ void AnalysisSupportHelpers::addMissingOutputsToBuilder(std::vector<InputSpec> c
   }
 }
 
+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 value = i.defaultValue.get<std::string>();
+        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));
+        }
+      }
+    }
+  }
+}
+
 // =============================================================================
 DataProcessorSpec AnalysisSupportHelpers::getOutputObjHistSink(ConfigContext const& ctx)
 {
diff --git a/Framework/Core/src/ArrowSupport.cxx b/Framework/Core/src/ArrowSupport.cxx
diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx
diff --git a/Framework/TestWorkflows/CMakeLists.txt b/Framework/TestWorkflows/CMakeLists.txt
diff --git a/Framework/TestWorkflows/src/o2TestAnalysisCCDB.cxx b/Framework/TestWorkflows/src/o2TestAnalysisCCDB.cxx
