WIP DPL Analysis: centralised CCDB support in analysis

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.

Author: ktf

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)
 

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&) final
+  {
+    return o2::framework::AnalysisCCDBHelpers::fetchFromCCDB();
+  }
+};
+
 DEFINE_DPL_PLUGINS_BEGIN
 DEFINE_DPL_PLUGIN_INSTANCE(CCDBFetcherPlugin, CustomAlgorithm);
+DEFINE_DPL_PLUGIN_INSTANCE(AnalysisCCDBFetcherPlugin, CustomAlgorithm);
 DEFINE_DPL_PLUGINS_END

Framework/Core/include/Framework/ASoA.h

@@ -2248,11 +2248,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 +2316,43 @@ 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_COLUMN(_Name_, _Getter_, _Type_) \
   DECLARE_SOA_COLUMN_FULL(_Name_, _Getter_, _Type_, "f" #_Name_)
 
@@ -3188,6 +3228,36 @@ 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>;                                                                             \
+    using ccdb_pack_t = framework::pack<__VA_ARGS__>;                                                                               \
+    /*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__)
 

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;

Framework/Core/include/Framework/AnalysisSupportHelpers.h

@@ -43,10 +43,13 @@ struct AnalysisSupportHelpers {
                                          std::vector<InputSpec>& requestedAODs,
                                          std::vector<InputSpec>& requestedDYNs,
                                          DataProcessorSpec& publisher);
+  static void addMissingOutputsToAnalysisCCDBFetcher(std::vector<OutputSpec> const& providedTimestampedCCDBObjecs,
+                                                     std::vector<InputSpec> const& requestedTimestampedCCDBObjects,
+                                                     DataProcessorSpec& publisher);
 
-  /// Match all inputs of kind ATSK and write them to a ROOT file,
-  /// one root file per originating task.
-  static DataProcessorSpec getOutputObjHistSink(ConfigContext const&);
+    /// Match all inputs of kind ATSK and write them to a ROOT file,
+    /// one root file per originating task.
+    static DataProcessorSpec getOutputObjHistSink(ConfigContext const&);
   /// writes inputs of kind AOD to file
   static DataProcessorSpec getGlobalAODSink(ConfigContext const&);
   /// Get the data director

Framework/Core/include/Framework/AnalysisTask.h

@@ -26,6 +26,7 @@
 #include "Framework/TypeIdHelpers.h"
 #include "Framework/ArrowTableSlicingCache.h"
 #include "Framework/AnalysisDataModel.h"
+#include <iostream>
 
 #include <arrow/compute/kernel.h>
 #include <arrow/table.h>
@@ -133,6 +134,7 @@ struct AnalysisDataProcessorBuilder {
   template <soa::is_table... As>
   static void addInputsAndExpressions(uint32_t hash, const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos)
   {
+    std::cout << "name: " << name << std::endl;
     int ai = -1;
     ([&ai, &hash, &eInfos, &name, &value, &inputs]() mutable {
       ++ai;

Framework/Core/src/AnalysisSupportHelpers.cxx

@@ -207,6 +207,34 @@ void AnalysisSupportHelpers::addMissingOutputsToBuilder(std::vector<InputSpec> c
   }
 }
 
+void AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher(
+                                                        std::vector<OutputSpec> const& providedTimestampedCDBObjects,
+                                                        std::vector<InputSpec> const& requestedTimestampedCCDBObjects,
+                                                        DataProcessorSpec& publisher)
+{
+  for (auto& input : requestedTimestampedCCDBObjects) {
+    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>();
+        std::cout << "XXX " << value << std::endl;
+    //    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)
 {

Framework/Core/src/ArrowSupport.cxx

@@ -19,6 +19,8 @@
 #include "Framework/ServiceRegistry.h"
 #include "Framework/ConfigContext.h"
 #include "Framework/CommonDataProcessors.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/DataSpecViews.h"
 #include "Framework/DeviceSpec.h"
 #include "Framework/EndOfStreamContext.h"
 #include "Framework/Tracing.h"
@@ -27,6 +29,7 @@
 #include "Framework/DeviceInfo.h"
 #include "Framework/DevicesManager.h"
 #include "Framework/DeviceConfig.h"
+#include "Framework/PluginManager.h"
 #include "Framework/ServiceMetricsInfo.h"
 #include "WorkflowHelpers.h"
 #include "Framework/WorkflowSpecNode.h"
@@ -441,13 +444,16 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
     .adjustTopology = [](WorkflowSpecNode& node, ConfigContext const& ctx) {
       auto& workflow = node.specs;
       auto spawner = std::find_if(workflow.begin(), workflow.end(), [](DataProcessorSpec const& spec) { return spec.name == "internal-dpl-aod-spawner"; });
+      auto analysisCCDB = std::find_if(workflow.begin(), workflow.end(), [](DataProcessorSpec const& spec) { return spec.name == "internal-dpl-aod-ccdb"; });
       auto builder = std::find_if(workflow.begin(), workflow.end(), [](DataProcessorSpec const& spec) { return spec.name == "internal-dpl-aod-index-builder"; });
       auto reader = std::find_if(workflow.begin(), workflow.end(), [](DataProcessorSpec const& spec) { return spec.name == "internal-dpl-aod-reader"; });
       auto writer = std::find_if(workflow.begin(), workflow.end(), [](DataProcessorSpec const& spec) { return spec.name == "internal-dpl-aod-writer"; });
       auto &ac = ctx.services().get<AnalysisContext>();
       ac.requestedAODs.clear();
       ac.requestedDYNs.clear();
       ac.providedDYNs.clear();
+      ac.providedTIMs.clear();
+      ac.requestedTIMs.clear();
 
 
       auto inputSpecLessThan = [](InputSpec const& lhs, InputSpec const& rhs) { return DataSpecUtils::describe(lhs) < DataSpecUtils::describe(rhs); };
@@ -511,6 +517,26 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         AnalysisSupportHelpers::addMissingOutputsToSpawner({}, ac.spawnerInputs, ac.requestedAODs, *spawner);
       }
 
+      if (analysisCCDB != workflow.end()) {
+        for (auto& d : workflow | views::exclude_by_name(analysisCCDB->name)) {
+          d.inputs | views::partial_match_filter(header::DataOrigin{"TIM"}) | sinks::update_input_list(ac.requestedTIMs);
+          d.outputs | views::partial_match_filter(header::DataOrigin{"TIM"}) | sinks::append_to(ac.providedTIMs);
+        }
+        std::sort(ac.requestedTIMs.begin(), ac.requestedTIMs.end(), inputSpecLessThan);
+        std::sort(ac.providedTIMs.begin(), ac.providedTIMs.end(), outputSpecLessThan);
+        // Use ranges::to<std::vector<>> in C++23...
+        ac.analysisCCDBInputs.clear();
+        ac.requestedTIMs | views::filter_not_matching(ac.providedTIMs) | sinks::append_to(ac.analysisCCDBInputs);
+
+        // recreate inputs and outputs
+        analysisCCDB->outputs.clear();
+        analysisCCDB->inputs.clear();
+        // replace AlgorithmSpec
+        // FIXME: it should be made more generic, so it does not need replacement...
+        analysisCCDB->algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkCCDBSupport", "AnalysisCCDBFetcherPlugin", ctx);
+        AnalysisSupportHelpers::addMissingOutputsToAnalysisCCDBFetcher(ac.providedTIMs, ac.requestedTIMs, *analysisCCDB);
+      }
+
       if (writer != workflow.end()) {
         workflow.erase(writer);
       }
@@ -538,6 +564,8 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         }
       }
 
+
+
       // replace writer as some outputs may have become dangling and some are now consumed
       auto [outputsInputs, isDangling] = WorkflowHelpers::analyzeOutputs(workflow);