DPL Analysis: remove compatibility with older FairMQ

Framework/CCDBSupport/src/AnalysisCCDBHelpers.cxx

@@ -111,33 +111,21 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
         auto it = ccdbUrls.find(m.name);
         fieldMetadata->Append("url", it != ccdbUrls.end() ? it->second : m.defaultValue.asString());
         auto columnName = m.name.substr(strlen("ccdb:"));
-#if (FAIRMQ_VERSION_DEC >= 111000)
         fields.emplace_back(std::make_shared<arrow::Field>(columnName, soa::asArrowDataType<int64_t[3]>(), false, fieldMetadata));
-#else
-        fields.emplace_back(std::make_shared<arrow::Field>(columnName, arrow::binary_view(), false, fieldMetadata));
-#endif
       }
       schemas.emplace_back(std::make_shared<arrow::Schema>(fields, schemaMetadata));
     }
 
-#if (FAIRMQ_VERSION_DEC >= 111000)
     std::vector<std::pair<uint32_t, std::shared_ptr<arrow::FixedSizeListBuilder>>> allbuilders;
-#else
-    std::vector<std::pair<uint32_t, std::shared_ptr<arrow::BinaryViewBuilder>>> allbuilders;
-#endif
     allbuilders.resize([&schemas]() { size_t size = 0; for (auto& schema : schemas) { size += schema->num_fields(); }; return size; }());
     auto* pool = arrow::default_memory_pool();
 
     int idx = 0;
     int sidx = 0;
     for (auto const& schema : schemas) {
       for (auto const& _ : schema->fields()) {
-#if (FAIRMQ_VERSION_DEC >= 111000)
         auto value_builder = std::make_shared<arrow::Int64Builder>();
         allbuilders[idx] = std::make_pair(sidx, std::make_shared<arrow::FixedSizeListBuilder>(pool, std::move(value_builder), 3));
-#else
-        allbuilders[idx] = std::make_pair(sidx, std::make_shared<arrow::BinaryViewBuilder>());
-#endif
         ++idx;
       }
       ++sidx;
@@ -227,16 +215,11 @@ AlgorithmSpec AnalysisCCDBHelpers::fetchFromCCDB(ConfigContext const& /*ctx*/)
                 lastId.value = response.id.value;
                 allocator.adoptFromCache(output, response.id, header::gSerializationMethodCCDB);
               }
-#if (FAIRMQ_VERSION_DEC >= 111000)
               result &= builder.second->Append();
               auto* value_builder = dynamic_cast<arrow::Int64Builder*>(builder.second->value_builder());
               result &= value_builder->Append(response.id.handle);
               result &= value_builder->Append(response.id.segment);
               result &= value_builder->Append(response.size);
-#else
-              char const* address = reinterpret_cast<char const*>(response.id.value);
-              result &= builder.second->Append(std::string_view(address, response.size));
-#endif
               ++bi;
             }
             if (!result.ok()) {

Framework/Core/include/Framework/ASoA.h

@@ -49,9 +49,7 @@ struct MetaHeader;
 namespace o2::framework
 {
 using ListVector = std::vector<std::vector<int64_t>>;
-#if (FAIRMQ_VERSION_DEC >= 111000)
 using PointerReconstructor = std::function<std::byte*(fair::mq::shmem::MetaHeader&&)>;
-#endif
 
 std::string cutString(std::string&& str);
 std::string strToUpper(std::string&& str);
@@ -1196,7 +1194,7 @@ struct TableIterator : IP, C... {
   {
     doSetCurrentInternal(internal_index_columns_t{}, table);
   }
-#if (FAIRMQ_VERSION_DEC >= 111000)
+
   void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
   {
     [&pointerReconstructor, this]<typename... Cs>(framework::pack<Cs...>) {
@@ -1210,7 +1208,6 @@ struct TableIterator : IP, C... {
        ...);
     }(all_columns{});
   }
-#endif
 
  private:
   /// Helper to move at the end of columns which actually have an iterator.
@@ -2185,12 +2182,12 @@ class Table
   {
     return self_t{mArrowTableRef.makeEmpty()};
   }
-#if (FAIRMQ_VERSION_DEC >= 111000)
+
   void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
   {
     mBegin.setPointerReconstructor(pointerReconstructor);
   }
-#endif
+
  private:
   template <typename T>
   arrow::ChunkedArray* lookupColumn()
@@ -2372,7 +2369,6 @@ consteval static std::string_view namespace_prefix()
   };                                                                                                                                                                              \
   [[maybe_unused]] static constexpr o2::framework::expressions::BindingNode _Getter_ { _Label_, _Name_::hash, o2::framework::expressions::selectArrowType<_Type_>() }
 
-#if (FAIRMQ_VERSION_DEC >= 111000)
 #define DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, _Label_, _Getter_, _ConcreteType_, _CCDBQuery_)                      \
   struct _Name_ : o2::soa::Column<int64_t[3], _Name_> {                                                           \
     static constexpr const char* mLabel = _Label_;                                                                \
@@ -2422,50 +2418,6 @@ consteval static std::string_view namespace_prefix()
       return _Getter_();                                                                                          \
     }                                                                                                             \
   };
-#else
-#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                                                                               \
-    {                                                                                                             \
-      if constexpr (std::same_as<_ConcreteType_, std::span<std::byte>>) {                                         \
-        return *mColumnIterator;                                                                                  \
-      } else {                                                                                                    \
-        static std::byte* payload = nullptr;                                                                      \
-        static _ConcreteType_* deserialised = nullptr;                                                            \
-        static TClass* c = TClass::GetClass(#_ConcreteType_);                                                     \
-        auto span = *mColumnIterator;                                                                             \
-        if (payload != (std::byte*)span.data()) {                                                                 \
-          payload = (std::byte*)span.data();                                                                      \
-          delete deserialised;                                                                                    \
-          TBufferFile f(TBufferFile::EMode::kRead, span.size(), (char*)span.data(), kFALSE);                      \
-          deserialised = (_ConcreteType_*)soa::extractCCDBPayload((char*)payload, span.size(), c, "ccdb_object"); \
-        }                                                                                                         \
-        return *deserialised;                                                                                     \
-      }                                                                                                           \
-    }                                                                                                             \
-                                                                                                                  \
-    decltype(auto)                                                                                                \
-      get() const                                                                                                 \
-    {                                                                                                             \
-      return _Getter_();                                                                                          \
-    }                                                                                                             \
-  };
-#endif
 
 #define DECLARE_SOA_CCDB_COLUMN(_Name_, _Getter_, _ConcreteType_, _CCDBQuery_) \
   DECLARE_SOA_CCDB_COLUMN_FULL(_Name_, "f" #_Name_, _Getter_, _ConcreteType_, _CCDBQuery_)
@@ -3778,12 +3730,12 @@ class FilteredBase : public T
   {
     return mCached;
   }
-#if (FAIRMQ_VERSION_DEC >= 111000)
+
   void setPointerReconstructor(framework::PointerReconstructor const& pointerReconstructor)
   {
     mFilteredBegin.setPointerReconstructor(pointerReconstructor);
   }
-#endif
+
  private:
   void resetRanges()
   {

Framework/Core/include/Framework/AnalysisDataModel.h

@@ -27,9 +27,7 @@
 #include "Framework/PID.h"
 
 #include <fairmq/Version.h>
-#if (FAIRMQ_VERSION_DEC >= 111000)
 #include <fairmq/shmem/Common.h>
-#endif
 
 namespace o2
 {

Framework/Core/include/Framework/AnalysisTask.h

@@ -304,19 +304,14 @@ struct AnalysisDataProcessorBuilder {
   }
 
   template <typename Task, is_table_iterator_or_enumeration Grouping, std::ranges::input_range R, soa::is_table... Associated>
-#if (FAIRMQ_VERSION_DEC >= 111000)
   static void invokeProcess(Task& task, InputRecord& inputs, R matchers, PointerReconstructor const& pointerReconstructor, void (Task::*processingFunction)(Grouping, Associated...), std::vector<ExpressionInfo>& infos, ArrowTableSlicingCache& slices, header::DataOrigin newOrigin = header::DataOrigin{"AOD"})
-#else
-  static void invokeProcess(Task& task, InputRecord& inputs, R matchers, void (Task::*processingFunction)(Grouping, Associated...), std::vector<ExpressionInfo>& infos, ArrowTableSlicingCache& slices, header::DataOrigin newOrigin = header::DataOrigin{"AOD"})
-#endif
   {
     using G = std::decay_t<Grouping>;
     auto groupingTable = AnalysisDataProcessorBuilder::bindGroupingTable(inputs, matchers, processingFunction, infos);
-#if (FAIRMQ_VERSION_DEC >= 111000)
+
     if constexpr (!is_enumeration<G>) {
       groupingTable.setPointerReconstructor(pointerReconstructor);
     }
-#endif
     constexpr const int numElements = nested_brace_constructible_size<false, std::decay_t<Task>>() / 10;
 
     // set filtered tables for partitions with grouping
@@ -357,12 +352,10 @@ struct AnalysisDataProcessorBuilder {
            ...);
         },
         associatedTables);
-#if (FAIRMQ_VERSION_DEC >= 111000)
       std::apply([&pointerReconstructor](auto&... table) {
         (table.setPointerReconstructor(pointerReconstructor), ...);
       },
                  associatedTables);
-#endif
 
       auto binder = [&task, &groupingTable, &associatedTables](auto& x) mutable {
         x.bindExternalIndices(&groupingTable, &std::get<std::decay_t<Associated>>(associatedTables)...);
@@ -393,12 +386,11 @@ struct AnalysisDataProcessorBuilder {
         auto slicer = GroupSlicer(groupingTable, associatedTables, slices, newOrigin);
         for (auto& slice : slicer) {
           auto associatedSlices = slice.associatedTables();
-#if (FAIRMQ_VERSION_DEC >= 111000)
           std::apply([&pointerReconstructor](auto&... table) {
             (table.setPointerReconstructor(pointerReconstructor), ...);
           },
                      associatedSlices);
-#endif
+
           overwriteInternalIndices(associatedSlices, associatedTables);
           std::apply(
             [&binder](auto&... x) mutable {
@@ -667,7 +659,6 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
       ic.services().get<ArrowTableSlicingCacheDef>().setCaches(std::move(bindingsKeys));
       ic.services().get<ArrowTableSlicingCacheDef>().setCachesUnsorted(std::move(bindingsKeysUnsorted));
       ic.services().get<ArrowTableSlicingCacheDef>().setOrigin(newOrigin);
-#if (FAIRMQ_VERSION_DEC >= 111000)
       PointerReconstructor pointerReconstructor(nullptr);
       bool hasCCDBTables = !ic.services().get<DanglingEdgesContext>().requestedTIMs.empty();
 
@@ -677,9 +668,6 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
           auto& spec = pc.services().get<DanglingEdgesContext>().requestedTIMs.front();
           pointerReconstructor = proxy.getShmPointerReconstructor(spec, 0);
         }
-#else
-      return [task, expressionInfos, inputInfos, newOrigin](ProcessingContext& pc) mutable {
-#endif
         // load the ccdb object from their cache
         homogeneous_apply_refs_sized<numElements>([&pc](auto& element) { return analysis_task_parsers::newDataframeCondition(pc.inputs(), element); }, *task.get());
         // reset partitions once per dataframe
@@ -704,28 +692,16 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
         if constexpr (requires { &T::process; }) {
           auto loc = std::ranges::find_if(inputInfos, [](auto const& info) { return info.hash == o2::framework::TypeIdHelpers::uniqueId<decltype(&T::process)>(); });
           auto matchers = loc == inputInfos.end() ? std::vector<std::pair<int, ConcreteDataMatcher>>{} : loc->matchers;
-#if (FAIRMQ_VERSION_DEC >= 111000)
           AnalysisDataProcessorBuilder::invokeProcess(*(task.get()), pc.inputs(), matchers, pointerReconstructor, &T::process, expressionInfos, slices, newOrigin);
-#else
-          AnalysisDataProcessorBuilder::invokeProcess(*(task.get()), pc.inputs(), matchers, &T::process, expressionInfos, slices, newOrigin);
-#endif
         }
         // execute optional process()
         homogeneous_apply_refs_sized<numElements>(
-#if (FAIRMQ_VERSION_DEC >= 111000)
           [&pc, &expressionInfos, &task, &slices, &inputInfos, &newOrigin, &pointerReconstructor](auto& x) {
-#else
-          [&pc, &expressionInfos, &task, &slices, &inputInfos, &newOrigin](auto& x) {
-#endif
             if constexpr (is_process_configurable<decltype(x)>) {
               if (x.value == true) {
                 auto loc = std::ranges::find_if(inputInfos, [](auto const& info) { return info.hash == o2::framework::TypeIdHelpers::uniqueId<decltype(x.process)>(); });
                 auto matchers = loc == inputInfos.end() ? std::vector<std::pair<int, ConcreteDataMatcher>>{} : loc->matchers;
-#if (FAIRMQ_VERSION_DEC >= 111000)
                 AnalysisDataProcessorBuilder::invokeProcess(*task.get(), pc.inputs(), matchers, pointerReconstructor, x.process, expressionInfos, slices, newOrigin);
-#else
-                AnalysisDataProcessorBuilder::invokeProcess(*task.get(), pc.inputs(), matchers, x.process, expressionInfos, slices, newOrigin);
-#endif
                 return true;
               }
               return false;

Framework/Core/include/Framework/DataAllocator.h

@@ -43,9 +43,7 @@
 // Do not change this for a full inclusion of fair::mq::Device.
 #include <fairmq/FwdDecls.h>
 #include <fairmq/Version.h>
-#if (FAIRMQ_VERSION_DEC >= 111000)
 #include <fairmq/shmem/Message.h>
-#endif
 
 namespace arrow
 {
@@ -635,11 +633,9 @@ DataAllocator::CacheId DataAllocator::adoptContainer(const Output& spec, Contain
     // clone is indistinguishable from the original, we can keep sending
     // the original.
     cacheId.value = context.addToCache(payloadMessage);
-#if (FAIRMQ_VERSION_DEC >= 111000)
     auto meta = dynamic_cast<fair::mq::shmem::Message*>(payloadMessage.get())->GetMeta();
     cacheId.handle = meta.fHandle;
     cacheId.segment = meta.fSegmentId;
-#endif
   }
 
   context.add<MessageContext::TrivialObject>(std::move(headerMessage), std::move(payloadMessage), routeIndex);

Framework/Core/include/Framework/FairMQDeviceProxy.h

@@ -21,9 +21,7 @@
 #include "Framework/ForwardRoute.h"
 #include <fairmq/FwdDecls.h>
 #include <fairmq/Version.h>
-#if (FAIRMQ_VERSION_DEC >= 111000)
 #include <fairmq/shmem/Common.h>
-#endif
 #include <vector>
 
 namespace o2::header
@@ -33,9 +31,8 @@ struct DataHeader;
 
 namespace o2::framework
 {
-#if (FAIRMQ_VERSION_DEC >= 111000)
 using PointerReconstructor = std::function<std::byte*(fair::mq::shmem::MetaHeader&&)>;
-#endif
+
 /// Helper class to hide fair::mq::Device headers in the DataAllocator header.
 /// This is done because fair::mq::Device brings in a bunch of boost.mpl /
 /// boost.fusion stuff, slowing down compilation times enourmously.
@@ -65,10 +62,8 @@ class FairMQDeviceProxy
   [[nodiscard]] ChannelIndex getForwardChannelIndexByName(std::string const& channelName) const;
   /// Retrieve the channel index from a given OutputSpec and the associated timeslice
   [[nodiscard]] ChannelIndex getOutputChannelIndex(OutputSpec const& spec, size_t timeslice) const;
-#if (FAIRMQ_VERSION_DEC >= 111000)
   /// Retrieve the pointer-reconstruction function for the shm manager for a given input spec
   [[nodiscard]] PointerReconstructor getShmPointerReconstructor(InputSpec const& spec, size_t timeslice);
-#endif
   /// Retrieve the channel index from a given OutputSpec and the associated timeslice
   void getMatchingForwardChannelIndexes(std::vector<ChannelIndex>& result, header::DataHeader const& header, size_t timeslice) const;
   /// ChannelIndex from a RouteIndex

Framework/Core/src/FairMQDeviceProxy.cxx

@@ -365,7 +365,6 @@ void FairMQDeviceProxy::bind(std::vector<OutputRoute> const& outputs, std::vecto
   mStateChangeCallback = newStatePending;
 }
 
-#if (FAIRMQ_VERSION_DEC >= 111000)
 PointerReconstructor FairMQDeviceProxy::getShmPointerReconstructor(InputSpec const& spec, size_t timeslice)
 {
   assert(mInputRoutes.size() == mInputs.size());
@@ -384,5 +383,4 @@ PointerReconstructor FairMQDeviceProxy::getShmPointerReconstructor(InputSpec con
   }
   return {};
 }
-#endif
 } // namespace o2::framework