[Framework] Add opt-in skipping of empty grouped slices

Filter empty associations before constructing typed slices while preserving
the original grouping policy and windows. Keep IncludeEmpty as the default
and document consumer and cache preconditions.

Add Arrow-backed integration coverage for Pair and Triple policies, iterator
copies, cache refresh, filtered grouping, bindings and window metadata.

Author: BongHwi

Framework/Core/CMakeLists.txt

@@ -239,6 +239,7 @@ add_executable(o2-test-framework-core
               test/test_FrameworkDataFlowToO2Control.cxx
               test/test_Graphviz.cxx
               test/test_GroupSlicer.cxx
+              test/test_GroupedCombinations.cxx
               test/test_HistogramRegistry.cxx
               test/test_HTTPParser.cxx
               test/test_IndexBuilder.cxx

Framework/Core/include/Framework/GroupedCombinations.h

@@ -21,6 +21,25 @@
 namespace o2::framework
 {
 
+/// Select which tuples of the original grouping policy are yielded.
+/// IncludeEmpty preserves the existing stream, including empty associated slices.
+/// SkipEmpty omits tuples with any empty association. Use it only when omitting
+/// those tuples has no observable effect on the consumer, including counters,
+/// normalization, window bookkeeping, random-number streams and diagnostics.
+/// Window flags and neighbour counts retain their original-policy meaning: they
+/// are not recomputed for surviving tuples, and a window may yield no tuple.
+/// The grouping policy must terminate and advance without observable side effects;
+/// an entirely empty association can end iteration without visiting every step.
+enum class GroupedSlicePolicy { IncludeEmpty,
+                                SkipEmpty };
+
+template <typename T>
+struct SupportsEmptyAssociatedGate : std::false_type {};
+
+template <aod::is_aod_hash L, aod::is_aod_hash D, aod::is_origin_hash O, typename... C>
+struct SupportsEmptyAssociatedGate<soa::Table<L, D, O, C...>>
+  : std::bool_constant<O::hash != "JOIN"_h && O::hash != "CONC"_h> {};
+
 // Create an instance of a tuple interleaved from given tuples
 template <typename... T1s, typename... T2s, std::size_t... Is>
 auto interleaveTuplesImpl(std::tuple<T1s...>& t1, std::tuple<T2s...>& t2, std::index_sequence<Is...>)
@@ -67,6 +86,12 @@ struct GroupedCombinationsGenerator {
       : GroupingPolicy(groupingPolicy), mIndexColumns{getMatchingIndexNode<G, As>()...}, cache{cache_}
     {
     }
+    GroupedIterator(const GroupingPolicy& groupingPolicy, SliceCache* cache_, GroupedSlicePolicy slicePolicy)
+      requires((SupportsEmptyAssociatedGate<As>::value && ...))
+      : GroupedIterator(groupingPolicy, cache_)
+    {
+      mSlicePolicy = slicePolicy;
+    }
     template <typename... T2s>
     GroupedIterator(const GroupingPolicy& groupingPolicy, const G& grouping, const std::tuple<T2s...>& associated, SliceCache* cache_)
       : GroupingPolicy(groupingPolicy),
@@ -172,8 +197,42 @@ struct GroupedCombinationsGenerator {
       return std::get<I>(*mAssociated).sliceByCached(mIndexColumns[I], ind, *cache);
     }
 
+    template <std::size_t I>
+    bool associatedHasRows() const
+    {
+      using A = std::tuple_element_t<I, std::tuple<As...>>;
+      auto const& associated = std::get<I>(*mAssociated);
+      if (associated.size() == 0) {
+        return false;
+      }
+      auto const& node = mIndexColumns[I];
+      auto localCache = cache->ptr->getCacheFor({"", soa::originReplacement(cache->ptr->newOrigin)(soa::getMatcherFromTypeForKey<A>(node.name)), node.name});
+      uint64_t index = *std::get<0>(std::get<I>(this->mCurrent).getIndices());
+      return localCache.getSliceFor(index).second != 0;
+    }
+
+    template <std::size_t... I>
+    bool allAssociatedHaveRows(std::index_sequence<I...>) const
+    {
+      return (associatedHasRows<I>() && ...);
+    }
+
     void setCurrentGroupedCombination()
     {
+      if constexpr ((SupportsEmptyAssociatedGate<As>::value && ...)) {
+        if (mSlicePolicy == GroupedSlicePolicy::SkipEmpty && !this->mIsEnd) {
+          // A globally empty association makes every original pair ineligible.
+          if (std::apply([](auto const&... tables) { return ((tables.size() == 0) || ...); }, *mAssociated)) {
+            GroupingPolicy::moveToEnd();
+            return;
+          }
+          // Advance the unchanged policy. Never remove grouping rows or refill
+          // a window with a more distant partner after rejecting an empty pair.
+          while (!this->mIsEnd && !allAssociatedHaveRows(std::index_sequence_for<As...>{})) {
+            this->addOne();
+          }
+        }
+      }
       if (!this->mIsEnd) {
         auto& currentGrouping = GroupingPolicy::mCurrent;
         mSlices.emplace(getAssociatedTables());
@@ -187,6 +246,7 @@ struct GroupedCombinationsGenerator {
     std::optional<std::tuple<As...>> mSlices;
     std::optional<GroupedIteratorType> mCurrentGrouped;
     SliceCache* cache = nullptr;
+    GroupedSlicePolicy mSlicePolicy = GroupedSlicePolicy::IncludeEmpty;
   };
 
   using iterator = GroupedIterator;
@@ -218,6 +278,24 @@ struct GroupedCombinationsGenerator {
   {
     setTables(grouping, associated);
   }
+  /// Opt in to the tuple filter described by GroupedSlicePolicy.
+  /// The type constraint accepts raw associated tables and excludes Join, Concat
+  /// and filtered associations. It does not validate sorting or cache contents.
+  /// As with sliceByCached, each matching association index must be sorted and
+  /// have an initialized sorted SliceCache entry for the current table/dataframe.
+  /// Grouping iterator indices must use the same coordinates as the cache keys.
+  /// The cache and bound tables must remain valid and unchanged during iteration.
+  /// Invalid/missing cache entries are not validated by this filter, and skipped
+  /// typed-slice construction need not reproduce diagnostics for invalid inputs.
+  template <typename... T2s>
+  GroupedCombinationsGenerator(const BP& binningPolicy, int catNeighbours, const T1& outsider, const G& grouping, const std::tuple<T2s...>& associated, SliceCache* cache, GroupedSlicePolicy slicePolicy)
+    requires((SupportsEmptyAssociatedGate<As>::value && ...))
+    : mBegin(GroupingPolicy(binningPolicy, catNeighbours, outsider), cache, slicePolicy),
+      mEnd(GroupingPolicy(binningPolicy, catNeighbours, outsider), cache, slicePolicy)
+  {
+    setTables(grouping, associated);
+  }
+
   GroupedCombinationsGenerator(GroupedCombinationsGenerator const&) = default;
   GroupedCombinationsGenerator& operator=(GroupedCombinationsGenerator const&) = default;
   ~GroupedCombinationsGenerator() = default;