DPL Analysis: prevent slice cache from updating unnecessarily (#14257)

* Cache setup now only happens after init when process configurables'
  values are final
* Add inline contrained functions to avoid using "overloaded"
* add error messages for unexpected situations

Author: aalkin

Framework/Core/include/Framework/ASoA.h

@@ -1400,10 +1400,10 @@ namespace o2::framework
 
 struct PreslicePolicyBase {
   const std::string binding;
-  StringPair bindingKey;
+  Entry bindingKey;
 
   bool isMissing() const;
-  StringPair const& getBindingKey() const;
+  Entry const& getBindingKey() const;
 };
 
 struct PreslicePolicySorted : public PreslicePolicyBase {
@@ -1428,7 +1428,7 @@ struct PresliceBase : public Policy {
   const std::string binding;
 
   PresliceBase(expressions::BindingNode index_)
-    : Policy{PreslicePolicyBase{{o2::soa::getLabelFromTypeForKey<T, OPT>(std::string{index_.name})}, std::make_pair(o2::soa::getLabelFromTypeForKey<T, OPT>(std::string{index_.name}), std::string{index_.name})}, {}}
+    : Policy{PreslicePolicyBase{{o2::soa::getLabelFromTypeForKey<T, OPT>(std::string{index_.name})}, Entry(o2::soa::getLabelFromTypeForKey<T, OPT>(std::string{index_.name}), std::string{index_.name})}, {}}
   {
   }
 
@@ -1508,7 +1508,7 @@ auto doSliceBy(T const* table, o2::framework::PresliceBase<C, Policy, OPT> const
 {
   if constexpr (OPT) {
     if (container.isMissing()) {
-      missingOptionalPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.second.c_str());
+      missingOptionalPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.key.c_str());
     }
   }
   uint64_t offset = 0;
@@ -1545,7 +1545,7 @@ auto doSliceBy(T const* table, o2::framework::PresliceBase<C, Policy, OPT> const
 {
   if constexpr (OPT) {
     if (container.isMissing()) {
-      missingOptionalPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.second.c_str());
+      missingOptionalPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.key.c_str());
     }
   }
   auto selection = container.getSliceFor(value);
@@ -1574,7 +1574,7 @@ auto doFilteredSliceBy(T const* table, o2::framework::PresliceBase<C, framework:
 {
   if constexpr (OPT) {
     if (container.isMissing()) {
-      missingOptionalPreslice(getLabelFromType<T>().data(), container.bindingKey.second.c_str());
+      missingOptionalPreslice(getLabelFromType<T>().data(), container.bindingKey.key.c_str());
     }
   }
   uint64_t offset = 0;

Framework/Core/include/Framework/AnalysisManagers.h

@@ -534,39 +534,43 @@ static void setGroupedCombination(C& comb, TG& grouping, std::tuple<Ts...>& asso
 /// Preslice handling
 template <typename T>
   requires(!is_preslice<T>)
-bool registerCache(T&, std::vector<StringPair>&, std::vector<StringPair>&)
+bool registerCache(T&, Cache&, Cache&)
 {
   return false;
 }
 
 template <is_preslice T>
   requires std::same_as<typename T::policy_t, framework::PreslicePolicySorted>
-bool registerCache(T& preslice, std::vector<StringPair>& bsks, std::vector<StringPair>&)
+bool registerCache(T& preslice, Cache& bsks, Cache&)
 {
   if constexpr (T::optional) {
     if (preslice.binding == "[MISSING]") {
       return true;
     }
   }
-  auto locate = std::find_if(bsks.begin(), bsks.end(), [&](auto const& entry) { return (entry.first == preslice.bindingKey.first) && (entry.second == preslice.bindingKey.second); });
+  auto locate = std::find_if(bsks.begin(), bsks.end(), [&](auto const& entry) { return (entry.binding == preslice.bindingKey.binding) && (entry.key == preslice.bindingKey.key); });
   if (locate == bsks.end()) {
     bsks.emplace_back(preslice.getBindingKey());
+  } else if (locate->enabled == false) {
+    locate->enabled = true;
   }
   return true;
 }
 
 template <is_preslice T>
   requires std::same_as<typename T::policy_t, framework::PreslicePolicyGeneral>
-bool registerCache(T& preslice, std::vector<StringPair>&, std::vector<StringPair>& bsksU)
+bool registerCache(T& preslice, Cache&, Cache& bsksU)
 {
   if constexpr (T::optional) {
     if (preslice.binding == "[MISSING]") {
       return true;
     }
   }
-  auto locate = std::find_if(bsksU.begin(), bsksU.end(), [&](auto const& entry) { return (entry.first == preslice.bindingKey.first) && (entry.second == preslice.bindingKey.second); });
+  auto locate = std::find_if(bsksU.begin(), bsksU.end(), [&](auto const& entry) { return (entry.binding == preslice.bindingKey.binding) && (entry.key == preslice.bindingKey.key); });
   if (locate == bsksU.end()) {
     bsksU.emplace_back(preslice.getBindingKey());
+  } else if (locate->enabled == false) {
+    locate->enabled = true;
   }
   return true;
 }

Framework/Core/include/Framework/AnalysisTask.h

@@ -65,21 +65,18 @@ concept is_enumeration = is_enumeration_v<std::decay_t<T>>;
 // the contents of an AnalysisTask...
 namespace {
 struct AnalysisDataProcessorBuilder {
-  template <typename G, typename... Args>
-  static void addGroupingCandidates(std::vector<StringPair>& bk, std::vector<StringPair>& bku)
+  template <soa::is_iterator G, typename... Args>
+  static void addGroupingCandidates(Cache& bk, Cache& bku, bool enabled)
   {
-    [&bk, &bku]<typename... As>(framework::pack<As...>) mutable {
-      std::string key;
-      if constexpr (soa::is_iterator<std::decay_t<G>>) {
-        key = std::string{"fIndex"} + o2::framework::cutString(soa::getLabelFromType<std::decay_t<G>>());
-      }
-      ([&bk, &bku, &key]() mutable {
+    [&bk, &bku, enabled]<typename... As>(framework::pack<As...>) mutable {
+      auto key = std::string{"fIndex"} + o2::framework::cutString(soa::getLabelFromType<std::decay_t<G>>());
+      ([&bk, &bku, &key, enabled]() mutable {
         if constexpr (soa::relatedByIndex<std::decay_t<G>, std::decay_t<As>>()) {
           auto binding = soa::getLabelFromTypeForKey<std::decay_t<As>>(key);
           if constexpr (o2::soa::is_smallgroups<std::decay_t<As>>) {
-            framework::updatePairList(bku, binding, key);
+            framework::updatePairList(bku, binding, key, enabled);
           } else {
-            framework::updatePairList(bk, binding, key);
+            framework::updatePairList(bk, binding, key, enabled);
           }
         }
       }(),
@@ -145,34 +142,72 @@ struct AnalysisDataProcessorBuilder {
   }
 
   /// helper to parse the process arguments
+  template <typename T>
+  inline static bool requestInputsFromArgs(T&, std::string const&, std::vector<InputSpec>&, std::vector<ExpressionInfo>&)
+  {
+    return false;
+  }
+  template <is_process_configurable T>
+  inline static bool requestInputsFromArgs(T& pc, std::string const& name, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eis)
+  {
+    AnalysisDataProcessorBuilder::inputsFromArgs(pc.process, (name + "/" + pc.name).c_str(), pc.value, inputs, eis);
+    return true;
+  }
+  template <typename T>
+  inline static bool requestCacheFromArgs(T&, Cache&, Cache&)
+  {
+    return false;
+  }
+  template <is_process_configurable T>
+  inline static bool requestCacheFromArgs(T& pc, Cache& bk, Cache& bku)
+  {
+    AnalysisDataProcessorBuilder::cacheFromArgs(pc.process, pc.value, bk, bku);
+    return true;
+  }
   /// 1. enumeration (must be the only argument)
   template <typename R, typename C, is_enumeration A>
-  static void inputsFromArgs(R (C::*)(A), const char* /*name*/, bool /*value*/, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>&, std::vector<StringPair>&, std::vector<StringPair>&)
+  static void inputsFromArgs(R (C::*)(A), const char* /*name*/, bool /*value*/, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>&) //, Cache&, Cache&)
   {
     std::vector<ConfigParamSpec> inputMetadata;
     // FIXME: for the moment we do not support begin, end and step.
     DataSpecUtils::updateInputList(inputs, InputSpec{"enumeration", "DPL", "ENUM", 0, Lifetime::Enumeration, inputMetadata});
   }
 
-  /// 2. grouping case - 1st argument is an iterator
+  /// 2. 1st argument is an iterator
   template <typename R, typename C, soa::is_iterator A, soa::is_table... Args>
-  static void inputsFromArgs(R (C::*)(A, Args...), const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos, std::vector<StringPair>& bk, std::vector<StringPair>& bku)
+  static void inputsFromArgs(R (C::*)(A, Args...), const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos) //, Cache& bk, Cache& bku)
     requires(std::is_lvalue_reference_v<A> && (std::is_lvalue_reference_v<Args> && ...))
   {
-    addGroupingCandidates<A, Args...>(bk, bku);
     constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<R (C::*)(A, Args...)>();
     addInputsAndExpressions<typename std::decay_t<A>::parent_t, Args...>(hash, name, value, inputs, eInfos);
   }
 
   /// 3. generic case
   template <typename R, typename C, soa::is_table... Args>
-  static void inputsFromArgs(R (C::*)(Args...), const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos, std::vector<StringPair>&, std::vector<StringPair>&)
+  static void inputsFromArgs(R (C::*)(Args...), const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos) //, Cache&, Cache&)
     requires(std::is_lvalue_reference_v<Args> && ...)
   {
     constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<R (C::*)(Args...)>();
     addInputsAndExpressions<Args...>(hash, name, value, inputs, eInfos);
   }
 
+  /// 1. enumeration (no grouping)
+  template <typename R, typename C, is_enumeration A>
+  static void cacheFromArgs(R (C::*)(A), bool, Cache&, Cache&)
+  {
+  }
+  /// 2. iterator (the only grouping case)
+  template <typename R, typename C, soa::is_iterator A, soa::is_table... Args>
+  static void cacheFromArgs(R (C::*)(A, Args...), bool value, Cache& bk, Cache& bku)
+  {
+    addGroupingCandidates<A, Args...>(bk, bku, value);
+  }
+  /// 3. generic case (no grouping)
+  template <typename R, typename C, soa::is_table A, soa::is_table... Args>
+  static void cacheFromArgs(R (C::*)(A, Args...), bool, Cache&, Cache&)
+  {
+  }
+
   template <soa::TableRef R>
   static auto extractTableFromRecord(InputRecord& record)
   {
@@ -480,8 +515,6 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
   std::vector<InputSpec> inputs;
   std::vector<ConfigParamSpec> options;
   std::vector<ExpressionInfo> expressionInfos;
-  std::vector<StringPair> bindingsKeys;
-  std::vector<StringPair> bindingsKeysUnsorted;
 
   /// make sure options and configurables are set before expression infos are created
   homogeneous_apply_refs([&options, &hash](auto& element) { return analysis_task_parsers::appendOption(options, element); }, *task.get());
@@ -490,23 +523,15 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
 
   /// parse process functions defined by corresponding configurables
   if constexpr (requires { &T::process; }) {
-    AnalysisDataProcessorBuilder::inputsFromArgs(&T::process, "default", true, inputs, expressionInfos, bindingsKeys, bindingsKeysUnsorted);
+    AnalysisDataProcessorBuilder::inputsFromArgs(&T::process, "default", true, inputs, expressionInfos);
   }
   homogeneous_apply_refs(
-    overloaded{
-      [name = name_str, &expressionInfos, &inputs, &bindingsKeys, &bindingsKeysUnsorted](framework::is_process_configurable auto& x) mutable {
-        // this pushes (argumentIndex,processHash,schemaPtr,nullptr) into expressionInfos for arguments that are Filtered/filtered_iterators
-        AnalysisDataProcessorBuilder::inputsFromArgs(x.process, (name + "/" + x.name).c_str(), x.value, inputs, expressionInfos, bindingsKeys, bindingsKeysUnsorted);
-        return true;
-      },
-      [](auto&) {
-        return false;
-      }},
+    [name = name_str, &expressionInfos, &inputs](auto& x) mutable {
+      // this pushes (argumentIndex, processHash, schemaPtr, nullptr) into expressionInfos for arguments that are Filtered/filtered_iterators
+      return AnalysisDataProcessorBuilder::requestInputsFromArgs(x, name, inputs, expressionInfos);
+    },
     *task.get());
 
-  // add preslice declarations to slicing cache definition
-  homogeneous_apply_refs([&bindingsKeys, &bindingsKeysUnsorted](auto& element) { return analysis_task_parsers::registerCache(element, bindingsKeys, bindingsKeysUnsorted); }, *task.get());
-
   // request base tables for spawnable extended tables and indices to be built
   // this checks for duplications
   homogeneous_apply_refs([&inputs](auto& element) {
@@ -526,7 +551,12 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
   requiredServices.insert(requiredServices.end(), arrowServices.begin(), arrowServices.end());
   homogeneous_apply_refs([&requiredServices](auto& element) { return analysis_task_parsers::addService(requiredServices, element); }, *task.get());
 
-  auto algo = AlgorithmSpec::InitCallback{[task = task, expressionInfos, bindingsKeys, bindingsKeysUnsorted](InitContext& ic) mutable {
+  auto algo = AlgorithmSpec::InitCallback{[task = task, expressionInfos](InitContext& ic) mutable {
+    Cache bindingsKeys;
+    Cache bindingsKeysUnsorted;
+    // add preslice declarations to slicing cache definition
+    homogeneous_apply_refs([&bindingsKeys, &bindingsKeysUnsorted](auto& element) { return analysis_task_parsers::registerCache(element, bindingsKeys, bindingsKeysUnsorted); }, *task.get());
+
     homogeneous_apply_refs([&ic](auto&& element) { return analysis_task_parsers::prepareOption(ic, element); }, *task.get());
     homogeneous_apply_refs([&ic](auto&& element) { return analysis_task_parsers::prepareService(ic, element); }, *task.get());
 
@@ -556,6 +586,16 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
       task->init(ic);
     }
 
+    /// parse process functions to enable requested grouping caches - note that at this state process configurables have their final values
+    if constexpr (requires { &T::process; }) {
+      AnalysisDataProcessorBuilder::cacheFromArgs(&T::process, true, bindingsKeys, bindingsKeysUnsorted);
+    }
+    homogeneous_apply_refs(
+      [&bindingsKeys, &bindingsKeysUnsorted](auto& x) mutable {
+        return AnalysisDataProcessorBuilder::requestCacheFromArgs(x, bindingsKeys, bindingsKeysUnsorted);
+      },
+      *task.get());
+
     ic.services().get<ArrowTableSlicingCacheDef>().setCaches(std::move(bindingsKeys));
     ic.services().get<ArrowTableSlicingCacheDef>().setCachesUnsorted(std::move(bindingsKeysUnsorted));
     // initialize global caches

Framework/Core/include/Framework/ArrowTableSlicingCache.h

@@ -34,51 +34,64 @@ struct SliceInfoUnsortedPtr {
   gsl::span<int64_t const> getSliceFor(int value) const;
 };
 
-using StringPair = std::pair<std::string, std::string>;
+struct Entry {
+  std::string binding;
+  std::string key;
+  bool enabled;
+
+  Entry(std::string b, std::string k, bool e = true)
+    : binding{b},
+      key{k},
+      enabled{e}
+  {
+  }
+};
+
+using Cache = std::vector<Entry>;
 
-void updatePairList(std::vector<StringPair>& list, std::string const& binding, std::string const& key);
+void updatePairList(Cache& list, std::string const& binding, std::string const& key, bool enabled);
 
 struct ArrowTableSlicingCacheDef {
   constexpr static ServiceKind service_kind = ServiceKind::Global;
-  std::vector<StringPair> bindingsKeys;
-  std::vector<StringPair> bindingsKeysUnsorted;
+  Cache bindingsKeys;
+  Cache bindingsKeysUnsorted;
 
-  void setCaches(std::vector<StringPair>&& bsks);
-  void setCachesUnsorted(std::vector<StringPair>&& bsks);
+  void setCaches(Cache&& bsks);
+  void setCachesUnsorted(Cache&& bsks);
 };
 
 struct ArrowTableSlicingCache {
   constexpr static ServiceKind service_kind = ServiceKind::Stream;
 
-  std::vector<StringPair> bindingsKeys;
+  Cache bindingsKeys;
   std::vector<std::shared_ptr<arrow::NumericArray<arrow::Int32Type>>> values;
   std::vector<std::shared_ptr<arrow::NumericArray<arrow::Int64Type>>> counts;
 
-  std::vector<StringPair> bindingsKeysUnsorted;
+  Cache bindingsKeysUnsorted;
   std::vector<std::vector<int>> valuesUnsorted;
   std::vector<ListVector> groups;
 
-  ArrowTableSlicingCache(std::vector<StringPair>&& bsks, std::vector<StringPair>&& bsksUnsorted = {});
+  ArrowTableSlicingCache(Cache&& bsks, Cache&& bsksUnsorted = {});
 
   // set caching information externally
-  void setCaches(std::vector<StringPair>&& bsks, std::vector<StringPair>&& bsksUnsorted = {});
+  void setCaches(Cache&& bsks, Cache&& bsksUnsorted = {});
 
   // update slicing info cache entry (assumes it is already present)
   arrow::Status updateCacheEntry(int pos, std::shared_ptr<arrow::Table> const& table);
   arrow::Status updateCacheEntryUnsorted(int pos, std::shared_ptr<arrow::Table> const& table);
 
   // helper to locate cache position
-  std::pair<int, bool> getCachePos(StringPair const& bindingKey) const;
-  int getCachePosSortedFor(StringPair const& bindingKey) const;
-  int getCachePosUnsortedFor(StringPair const& bindingKey) const;
+  std::pair<int, bool> getCachePos(Entry const& bindingKey) const;
+  int getCachePosSortedFor(Entry const& bindingKey) const;
+  int getCachePosUnsortedFor(Entry const& bindingKey) const;
 
   // get slice from cache for a given value
-  SliceInfoPtr getCacheFor(StringPair const& bindingKey) const;
-  SliceInfoUnsortedPtr getCacheUnsortedFor(StringPair const& bindingKey) const;
+  SliceInfoPtr getCacheFor(Entry const& bindingKey) const;
+  SliceInfoUnsortedPtr getCacheUnsortedFor(Entry const& bindingKey) const;
   SliceInfoPtr getCacheForPos(int pos) const;
   SliceInfoUnsortedPtr getCacheUnsortedForPos(int pos) const;
 
-  static void validateOrder(StringPair const& bindingKey, std::shared_ptr<arrow::Table> const& input);
+  static void validateOrder(Entry const& bindingKey, std::shared_ptr<arrow::Table> const& input);
 };
 } // namespace o2::framework
 

Framework/Core/include/Framework/GroupSlicer.h

@@ -55,7 +55,7 @@ struct GroupSlicer {
     {
       constexpr auto index = framework::has_type_at_v<std::decay_t<T>>(associated_pack_t{});
       auto binding = o2::soa::getLabelFromTypeForKey<std::decay_t<T>>(mIndexColumnName);
-      auto bk = std::make_pair(binding, mIndexColumnName);
+      auto bk = Entry(binding, mIndexColumnName);
       if constexpr (!o2::soa::is_smallgroups<std::decay_t<T>>) {
         if (table.size() == 0) {
           return;

Framework/Core/src/ASoA.cxx

@@ -194,7 +194,7 @@ bool PreslicePolicyBase::isMissing() const
   return binding == "[MISSING]";
 }
 
-StringPair const& PreslicePolicyBase::getBindingKey() const
+Entry const& PreslicePolicyBase::getBindingKey() const
 {
   return bindingKey;
 }