Merge be4cb11 into sapling-pr-archive-ktf

ktf · web-flow · commit e9cf89148339 · 2025-08-15T08:56:09.000+02:00
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&) 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
diff --git a/Framework/Core/include/Framework/ASoA.h b/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__)
 
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/AnalysisManagers.h b/Framework/Core/include/Framework/AnalysisManagers.h
@@ -159,7 +159,9 @@ const char* controlOption()
 }
 
 template <typename T>
-  requires(is_spawns<T> || is_builds<T> || is_defines<T>)
+concept with_base_table = requires(T const& t) { t.base_specs(); };
+
+template <with_base_table T>
 bool requestInputs(std::vector<InputSpec>& inputs, T const& entity)
 {
   auto base_specs = entity.base_specs();
diff --git a/Framework/Core/include/Framework/AnalysisSupportHelpers.h b/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
diff --git a/Framework/Core/include/Framework/AnalysisTask.h b/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;
diff --git a/Framework/Core/include/Framework/DataSpecViews.h b/Framework/Core/include/Framework/DataSpecViews.h
@@ -18,8 +18,50 @@ namespace o2::framework::views
 {
 static auto partial_match_filter(auto what)
 {
-  return std::views::filter([&what](auto const& t) -> bool { return DataSpecUtils::partialMatch(t, what); });
+  return std::views::filter([what](auto const& t) -> bool { return DataSpecUtils::partialMatch(t, what); });
 }
+
+static auto exclude_by_name(std::string name)
+{
+  return std::views::filter([name](auto const& t) -> bool { return t.name != name; });
+}
+
+static auto filter_not_matching(auto const& provided)
+{
+  return std::views::filter([&provided](auto const& input) { return std::none_of(provided.begin(), provided.end(), [&input](auto const& output) { return DataSpecUtils::match(input, output); }); });
+}
+
 } // namespace o2::framework::views
+//
+namespace o2::framework::sinks
+{
+template <class Container>
+struct append_to {
+  Container& c;
+  // ends the pipeline, returns the container
+  template <std::ranges::input_range R>
+  friend Container& operator|(R&& r, append_to self)
+  {
+    std::ranges::copy(r, std::back_inserter(self.c));
+    return self.c;
+  }
+};
+
+template <class Container>
+struct update_input_list {
+  Container& c;
+  // ends the pipeline, returns the container
+  template <std::ranges::input_range R>
+  friend Container& operator|(R&& r, update_input_list self)
+  {
+    for (auto& item : r) {
+      auto copy = item;
+      DataSpecUtils::updateInputList(self.c, std::move(copy));
+    }
+    return self.c;
+  }
+};
+
+} // namespace o2::framework::sinks
 
 #endif // O2_FRAMEWORK_DATASPECVIEWS_H_
diff --git a/Framework/Core/src/AnalysisSupportHelpers.cxx b/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)
 {
diff --git a/Framework/Core/src/ArrowSupport.cxx b/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);
 
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

Original file line number	Diff line number	Diff line change
`@@ -159,7 +159,9 @@ const char* controlOption()`
`159`	`159`	`}`
`160`	`160`
`161`	`161`	`template <typename T>`
`162`		`- requires(is_spawns<T> \|\| is_builds<T> \|\| is_defines<T>)`
	`162`	`+concept with_base_table = requires(T const& t) { t.base_specs(); };`
	`163`	`+`
	`164`	`+template <with_base_table T>`
`163`	`165`	`bool requestInputs(std::vector<InputSpec>& inputs, T const& entity)`
`164`	`166`	`{`
`165`	`167`	`auto base_specs = entity.base_specs();`