[RDF] Share column readers for Defines not affected by variations.

When a Define is added to a branch that gets varied, a new column reader
was created for every variation and for every slot, irrespective of
whether the column gets varied or not.
For snapshot with variations, this column would be written for every
variation, although it always evaluates to the same value. To prevent this
duplication, the column readers that point to the same Define are shared
after this commit. The readers are still cloned per slot though, so the
Define's value caches remain thread safe. This allows
SnapshotWithVariations to detect that columns are identical, so they are
not written multiple times.
The memory savings amount to 24 bytes per suppressed column reader,so
they won't have a notable impact.

Author: hageboeck

tree/dataframe/inc/ROOT/RDF/RDefineReader.hxx

@@ -62,7 +62,7 @@ class RDefinesWithReaders {
    // (see BookDefineJit). it is never null.
    std::shared_ptr<ROOT::Detail::RDF::RDefineBase> fDefine;
    // Column readers per variation (in the map) per slot (in the vector).
-   std::vector<std::unordered_map<std::string_view, std::unique_ptr<RDefineReader>>> fReadersPerVariation;
+   std::vector<std::unordered_map<std::string_view, std::shared_ptr<RDefineReader>>> fReadersPerVariation;
 
    // Strings that were already used to represent column names in this RDataFrame instance.
    ROOT::Internal::RDF::RStringCache &fCachedColNames;

tree/dataframe/src/RDefineReader.cxx

@@ -32,19 +32,28 @@ ROOT::Internal::RDF::RDefinesWithReaders::GetReader(unsigned int slot, std::stri
    if (it != defineReaders.end())
       return *it->second;
 
+   std::shared_ptr<RDefineReader> readerToReturn;
    auto *define = fDefine.get();
-   if (*nameIt != "nominal")
-      define = &define->GetVariedDefine(std::string(variationName));
-
-#if !defined(__clang__) && __GNUC__ >= 7 && __GNUC_MINOR__ >= 3
-   const auto insertion =
-      defineReaders.insert({*nameIt, std::make_unique<ROOT::Internal::RDF::RDefineReader>(slot, *define)});
-   return *insertion.first->second;
-#else
-   // gcc < 7.3 has issues with passing the non-movable std::pair temporary into the insert call
-   auto reader = std::make_unique<ROOT::Internal::RDF::RDefineReader>(slot, *define);
-   auto &ret = *reader;
-   defineReaders[*nameIt] = std::move(reader);
-   return ret;
-#endif
+   if (*nameIt == "nominal") {
+      readerToReturn = std::make_shared<RDefineReader>(slot, *define);
+   } else {
+      auto *variedDefine = &define->GetVariedDefine(std::string(variationName));
+      if (variedDefine == define) {
+         // The column in not affected by variations. We can return the same reader as for nominal
+         if (auto nominalReaderIt = defineReaders.find("nominal"); nominalReaderIt != defineReaders.end()) {
+            readerToReturn = nominalReaderIt->second;
+         } else {
+            // The nominal reader doesn't exist yet
+            readerToReturn = std::make_shared<RDefineReader>(slot, *define);
+            auto nominalNameIt = fCachedColNames.Insert("nominal");
+            defineReaders.insert({*nominalNameIt, readerToReturn});
+         }
+      } else {
+         readerToReturn = std::make_shared<RDefineReader>(slot, *variedDefine);
+      }
+   }
+
+   defineReaders.insert({*nameIt, readerToReturn});
+
+   return *readerToReturn;
 }