RNTupleMerger.cxx

Go to the documentation of this file.

/// \file RNTupleMerger.cxx
/// \author Jakob Blomer <jblomer@cern.ch>, Max Orok <maxwellorok@gmail.com>, Alaettin Serhan Mete <amete@anl.gov>,
/// Giacomo Parolini <giacomo.parolini@cern.ch>
/// \date 2020-07-08
/// \warning This is part of the ROOT 7 prototype! It will
/// change without notice. It might trigger earthquakes. Feedback is welcome!
 
/*************************************************************************
 * Copyright (C) 1995-2020, Rene Brun and Fons Rademakers.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
 *************************************************************************/
 
#include <ROOT/RError.hxx>
#include <ROOT/RNTuple.hxx>
#include <ROOT/RNTupleDescriptor.hxx>
#include <ROOT/RNTupleMerger.hxx>
#include <ROOT/RNTupleModel.hxx>
#include <ROOT/RNTupleTypes.hxx>
#include <ROOT/RNTupleUtils.hxx>
#include <ROOT/RPageAllocator.hxx>
#include <ROOT/RPageStorageFile.hxx>
#include <ROOT/RPageStorage.hxx>
#include <ROOT/RClusterPool.hxx>
#include <ROOT/RNTupleSerialize.hxx>
#include <ROOT/RNTupleZip.hxx>
#include <ROOT/RColumnElementBase.hxx>
#include <ROOT/RNTupleAttrUtils.hxx>
#include <TROOT.h>
#include <TFileMergeInfo.h>
#include <TFile.h>
#include <TKey.h>
 
#include <algorithm>
#include <deque>
#include <initializer_list>
#include <unordered_map>
#include <vector>
 
using ROOT::ENTupleColumnType;
using ROOT::RNTupleModel;
using ROOT::Internal::MakeUninitArray;
using ROOT::Internal::RCluster;
using ROOT::Internal::RColumnElementBase;
using ROOT::Internal::RNTupleSerializer;
using ROOT::Internal::RPageSink;
using ROOT::Internal::RPageSource;
using ROOT::Internal::RPageSourceFile;
using ROOT::Internal::RPageStorage;
 
using namespace ROOT::Internal;
 
static ROOT::RLogChannel &NTupleMergeLog()
{
   static ROOT::RLogChannel sLog("ROOT.NTuple.Merge");
   return sLog;
}
 
// TFile options parsing
// -------------------------------------------------------------------------------------
static bool BeginsWithDelimitedWord(const TString &str, const char *word)
{
   const Ssiz_t wordLen = strlen(word);
   if (str.Length() < wordLen)
      return false;
   if (!str.BeginsWith(word, TString::ECaseCompare::kIgnoreCase))
      return false;
   return str.Length() == wordLen || str(wordLen) == ' ';
}
 
template <typename T>
static std::optional<T> ParseStringOption(const TString &opts, const char *pattern,
                                          std::initializer_list<std::pair<const char *, T>> validValues)
{
   const Ssiz_t patternLen = strlen(pattern);
   assert(pattern[patternLen - 1] == '='); // we want to parse options with the format `option=Value`
   if (auto idx = opts.Index(pattern, 0, TString::ECaseCompare::kIgnoreCase);
       idx >= 0 && opts.Length() > idx + patternLen) {
      auto sub = TString(opts(idx + patternLen, opts.Length() - idx - patternLen));
      for (const auto &[name, value] : validValues) {
         if (BeginsWithDelimitedWord(sub, name)) {
            return value;
         }
      }
   }
   return std::nullopt;
}
 
static std::optional<ENTupleMergingMode> ParseOptionMergingMode(const TString &opts)
{
   return ParseStringOption<ENTupleMergingMode>(opts, "rntuple.MergingMode=",
                                                {
                                                   {"Filter", ENTupleMergingMode::kFilter},
                                                   {"Union", ENTupleMergingMode::kUnion},
                                                   {"Strict", ENTupleMergingMode::kStrict},
                                                });
}
 
static std::optional<ENTupleMergeErrBehavior> ParseOptionErrBehavior(const TString &opts)
{
   return ParseStringOption<ENTupleMergeErrBehavior>(opts, "rntuple.ErrBehavior=",
                                                     {
                                                        {"Abort", ENTupleMergeErrBehavior::kAbort},
                                                        {"Skip", ENTupleMergeErrBehavior::kSkip},
                                                     });
}
 
static std::optional<ENTupleMergeVersionBehavior> ParseOptionVersionBehavior(const TString &opts)
{
   return ParseStringOption<ENTupleMergeVersionBehavior>(
      opts, "rntuple.VersionBehavior=",
      {
         {"WarnOnHigherVersion", ENTupleMergeVersionBehavior::kWarnOnHigherVersion},
         {"AbortOnHigherVersion", ENTupleMergeVersionBehavior::kAbortOnHigherVersion},
      });
}
// -------------------------------------------------------------------------------------
 
// Entry point for TFileMerger. Internally calls RNTupleMerger::Merge().
Long64_t ROOT::RNTuple::Merge(TCollection *inputs, TFileMergeInfo *mergeInfo)
// IMPORTANT: this function must not throw, as it is used in exception-unsafe code (TFileMerger).
try {
   // Check the inputs
   if (!inputs || inputs->GetEntries() < 3 || !mergeInfo) {
      R__LOG_ERROR(NTupleMergeLog()) << "Invalid inputs.";
      return -1;
   }
 
   // Parse the input parameters
   TIter itr(inputs);
 
   // First entry is the RNTuple name
   std::string ntupleName = std::string(itr()->GetName());
 
   // Second entry is the output file
   TObject *secondArg = itr();
   TFile *outFile = dynamic_cast<TFile *>(secondArg);
   if (!outFile) {
      R__LOG_ERROR(NTupleMergeLog()) << "Second input parameter should be a TFile, but it's a "
                                     << secondArg->ClassName() << ".";
      return -1;
   }
 
   // Check if the output file already has a key with that name
   TKey *outKey = outFile->FindKey(ntupleName.c_str());
   ROOT::RNTuple *outNTuple = nullptr;
   if (outKey) {
      outNTuple = outKey->ReadObject<ROOT::RNTuple>();
      if (!outNTuple) {
         R__LOG_ERROR(NTupleMergeLog()) << "Output file already has key, but not of type RNTuple!";
         return -1;
      }
      // In principle, we should already be working on the RNTuple object from the output file, but just continue with
      // pointer we just got.
   }
 
   const bool defaultComp = mergeInfo->fOptions.Contains("DefaultCompression");
   const bool firstSrcComp = mergeInfo->fOptions.Contains("FirstSrcCompression");
   const bool extraVerbose = mergeInfo->fOptions.Contains("rntuple.ExtraVerbose");
   if (defaultComp && firstSrcComp) {
      // this should never happen through hadd, but a user may call RNTuple::Merge() from custom code.
      R__LOG_WARNING(NTupleMergeLog()) << "Passed both options \"DefaultCompression\" and \"FirstSrcCompression\": "
                                          "only the latter will apply.";
   }
   std::optional<std::uint32_t> compression;
   if (firstSrcComp) {
      // user passed -ff or -fk: use the same compression as the first RNTuple we find in the sources.
      // (do nothing here, the compression will be fetched below)
   } else if (!defaultComp) {
      // compression was explicitly passed by the user: use it.
      compression = outFile->GetCompressionSettings();
   } else {
      // user passed no compression-related options: use default
      compression = RCompressionSetting::EDefaults::kUseGeneralPurpose;
      R__LOG_INFO(NTupleMergeLog()) << "Using the default compression: " << *compression;
   }
 
   // The remaining entries are the input files
   std::vector<std::unique_ptr<RPageSourceFile>> sources;
   std::vector<RPageSource *> sourcePtrs;
 
   while (const auto &pitr = itr()) {
      TFile *inFile = dynamic_cast<TFile *>(pitr);
      ROOT::RNTuple *anchor = inFile ? inFile->Get<ROOT::RNTuple>(ntupleName.c_str()) : nullptr;
      if (!anchor) {
         R__LOG_INFO(NTupleMergeLog()) << "No RNTuple anchor named '" << ntupleName << "' from file '"
                                       << inFile->GetName() << "'";
         continue;
      }
 
      auto source = RPageSourceFile::CreateFromAnchor(*anchor);
      if (!compression) {
         // Get the compression of this RNTuple and use it as the output compression.
         // We currently assume all column ranges have the same compression, so we just peek at the first one.
         source->Attach(RNTupleSerializer::EDescriptorDeserializeMode::kRaw);
         DescriptorId_t firstClusterId = kInvalidDescriptorId;
         auto descGuard = source->FindClusterId(0, firstClusterId);
         if (firstClusterId == kInvalidDescriptorId) {
            R__LOG_ERROR(NTupleMergeLog())
               << "Asked to use the first source's compression as the output compression, but the "
                  "first source (file '"
               << inFile->GetName()
               << "') has an empty RNTuple, therefore the output compression could not be "
                  "determined.";
            return -1;
         }
         const auto &firstCluster = descGuard->GetClusterDescriptor(firstClusterId);
         auto colRangeIter = firstCluster.GetColumnRangeIterable();
         auto firstColRange = colRangeIter.begin();
         if (firstColRange == colRangeIter.end()) {
            R__LOG_ERROR(NTupleMergeLog())
               << "Asked to use the first source's compression as the output compression, but the "
                  "first source (file '"
               << inFile->GetName()
               << "') has an empty RNTuple, therefore the output compression could not be "
                  "determined.";
            return -1;
         }
         compression = (*firstColRange).GetCompressionSettings();
         R__LOG_INFO(NTupleMergeLog()) << "Using the first RNTuple's compression: " << *compression;
      }
      sources.push_back(std::move(source));
   }
 
   RNTupleWriteOptions writeOpts;
   assert(compression);
   writeOpts.SetCompression(*compression);
   auto destination = std::make_unique<ROOT::Internal::RPageSinkFile>(ntupleName, *outFile, writeOpts);
   std::unique_ptr<ROOT::RNTupleModel> model;
   // If we already have an existing RNTuple, copy over its descriptor to support incremental merging
   if (outNTuple) {
      auto outSource = RPageSourceFile::CreateFromAnchor(*outNTuple);
      outSource->Attach(RNTupleSerializer::EDescriptorDeserializeMode::kForWriting);
      outSource->LoadAllPageLists();
      auto desc = outSource->GetSharedDescriptorGuard();
      model = destination->InitFromDescriptor(desc.GetConstRef(), true /* copyClusters */);
   }
 
   // Interface conversion
   sourcePtrs.reserve(sources.size());
   for (const auto &s : sources) {
      sourcePtrs.push_back(s.get());
   }
 
   // Now merge
   RNTupleMerger merger{std::move(destination), std::move(model)};
   RNTupleMergeOptions mergerOpts;
   mergerOpts.fCompressionSettings = compression;
   mergerOpts.fExtraVerbose = extraVerbose;
   if (auto mergingMode = ParseOptionMergingMode(mergeInfo->fOptions)) {
      mergerOpts.fMergingMode = *mergingMode;
   }
   if (auto errBehavior = ParseOptionErrBehavior(mergeInfo->fOptions)) {
      mergerOpts.fErrBehavior = *errBehavior;
   }
   if (auto versionBehavior = ParseOptionVersionBehavior(mergeInfo->fOptions)) {
      mergerOpts.fVersionBehavior = *versionBehavior;
   }
   merger.Merge(sourcePtrs, mergerOpts).ThrowOnError();
 
   // Provide the caller with a merged anchor object (even though we've already
   // written it).
   *this = *outFile->Get<ROOT::RNTuple>(ntupleName.c_str());
 
   return 0;
} catch (const std::exception &ex) {
   R__LOG_ERROR(NTupleMergeLog()) << "Exception thrown while merging: " << ex.what();
   return -1;
}
 
// =========================================================================================
//
// Local types and functions
//
// =========================================================================================
namespace {
// Functor used to change the compression of a page to `fCompressionSettings`.
struct RChangeCompressionFunc {
   const RColumnElementBase &fSrcColElement;
   std::uint32_t fCompressionSettings;
   RPageStorage::RSealedPage &fSealedPage;
   ROOT::Internal::RPageAllocator &fPageAlloc;
   std::byte *fBuffer;
   std::size_t fBufSize;
   const ROOT::RNTupleWriteOptions &fWriteOpts;
 
   void operator()() const
   {
      fSealedPage.VerifyChecksumIfEnabled().ThrowOnError();
 
      const auto bytesPacked = fSrcColElement.GetPackedSize(fSealedPage.GetNElements());
      // TODO: this buffer could be kept and reused across pages
      std::unique_ptr<std::byte[]> unzipBufOwned;
      std::byte *unzipBuf;
      if (fCompressionSettings != 0) {
         unzipBufOwned = MakeUninitArray<std::byte>(bytesPacked);
         unzipBuf = unzipBufOwned.get();
      } else {
         unzipBuf = fBuffer;
      }
      ROOT::Internal::RNTupleDecompressor::Unzip(fSealedPage.GetBuffer(), fSealedPage.GetDataSize(), bytesPacked,
                                                 unzipBuf);
 
      const auto checksumSize = fWriteOpts.GetEnablePageChecksums() * sizeof(std::uint64_t);
      std::size_t nBytesZipped;
      if (fCompressionSettings != 0) {
         assert(fBuffer != unzipBuf);
         assert(fBufSize >= bytesPacked + checksumSize);
         nBytesZipped = ROOT::Internal::RNTupleCompressor::Zip(unzipBuf, bytesPacked, fCompressionSettings, fBuffer);
      } else {
         nBytesZipped = bytesPacked;
      }
      fSealedPage = {fBuffer, nBytesZipped + checksumSize, fSealedPage.GetNElements(), fSealedPage.GetHasChecksum()};
      fSealedPage.ChecksumIfEnabled();
   }
};
 
struct RTaskVisitor {
   std::optional<ROOT::Experimental::TTaskGroup> &fGroup;
 
   template <typename T>
   void operator()(T &&f)
   {
      if (fGroup)
         fGroup->Run(f);
      else
         f();
   }
};
 
struct RCommonField {
   const ROOT::RFieldDescriptor *fSrc;
   const ROOT::RFieldDescriptor *fDst;
 
   RCommonField(const ROOT::RFieldDescriptor &src, const ROOT::RFieldDescriptor &dst) : fSrc(&src), fDst(&dst) {}
};
 
/// Maps a column representation from a source to a destination RNTuple.
/// fSource and fDest are the first representation indices of a specific column.
///
/// When we merge fields from different RNTuples, two compatible fields may use different column
/// representations. When merging their columns we need to make sure that we keep the output
/// representation coherent, which is what this mapping is here for.
struct RColReprMapping {
   std::uint32_t fSource;
   std::uint32_t fDest;
};
 
/// A column extension that needs to be added to an output field.
/// Note that this also adds a mapping for each new representation, which is why it inherits RColReprMapping.
struct RColReprExtension : RColReprMapping {
   /// The new representations to be added
   std::vector<ROOT::Internal::RColumnFormat> fSourceRepr;
   /// The first element index that this column had in its source. When adding this representation to the destination,
   /// the new column will add this amount to the first element index of the 0th-representation column in the
   /// destination's current cluster.
   std::uint32_t fOrigFirstElementIndex = 0;
};
 
static std::optional<std::uint32_t>
FindColumnReprMapping(const std::vector<RColReprMapping> &mappings, std::uint32_t sourceReprIndex)
{
   for (const auto [src, dst] : mappings)
      if (src == sourceReprIndex)
         return dst;
   return std::nullopt;
}
 
struct RColumnOutInfo {
   ROOT::DescriptorId_t fColumnId = ROOT::kInvalidDescriptorId;
};
 
// { ".fully.qualified.fieldName.colInputIndex.colOutputReprIndex" => colOutputInfo }
using ColumnIdMap_t = std::unordered_map<std::string, RColumnOutInfo>;
 
struct RColumnMergeInfo {
   // This column name is built as a dot-separated concatenation of the ancestry of
   // the columns' parent fields' names plus the index of the column itself.
   // e.g. "Muon.pt.x._0"
   std::string fColumnName;
   // The column id in the source RNTuple
   ROOT::DescriptorId_t fInputId = ROOT::kInvalidDescriptorId;
   // The corresponding column id in the destination RNTuple (the mapping happens in AddColumnsFromField())
   ROOT::DescriptorId_t fOutputId = ROOT::kInvalidDescriptorId;
   std::uint16_t fOutputReprIndex = 0;
   // If nullopt, use the default in-memory type
   std::optional<std::type_index> fInMemoryType;
   const ROOT::RFieldDescriptor *fParentFieldDescriptor = nullptr;
   const ROOT::RNTupleDescriptor *fParentNTupleDescriptor = nullptr;
};
 
struct RSealedPageMergeData {
   // We use a std::deque so that references to the contained SealedPageSequence_t, and its iterators, are
   // never invalidated.
   std::deque<RPageStorage::SealedPageSequence_t> fPagesV;
   std::vector<RPageStorage::RSealedPageGroup> fGroups;
   std::vector<std::unique_ptr<std::byte[]>> fBuffers;
};
 
struct RColumnInfoGroup {
   std::vector<RColumnMergeInfo> fExtraDstColumns;
   std::vector<RColumnMergeInfo> fCommonColumns;
};
 
template <typename T>
using FieldCollectionMap_t = std::unordered_map<const ROOT::RFieldDescriptor *, std::vector<T>>;
 
struct RDescriptorsComparison {
   std::vector<const ROOT::RFieldDescriptor *> fExtraDstFields;
   std::vector<const ROOT::RFieldDescriptor *> fExtraSrcFields;
   std::vector<RCommonField> fCommonFields;
   // { parentId => subfields }
   // This is currently only used for the _userData attribute field.
   std::unordered_map<ROOT::DescriptorId_t, std::vector<const ROOT::RFieldDescriptor *>> fExtraSrcSubfields;
 
   // Empty if and only if the source can be fast merged.
   std::string fFastMergeErrors;
 
   // For each field that has more than 1 column representation in the output model,
   // maps the column representatives of the source field with those of the destination.
   // The key is the destination field.
   FieldCollectionMap_t<RColReprMapping> fColReprMappings;
   FieldCollectionMap_t<RColReprExtension> fColReprExtensions;
};
 
} // namespace
 
// Subprocedure of CompareDescriptorStructure, extracted for readability.
// Given two fields, attempts to match their column representations and schedules column extensions if necessary.
static void MatchColumnRepresentations(const ROOT::RNTupleDescriptor &srcDesc, const ROOT::RNTupleDescriptor &dstDesc,
                                       const ROOT::RFieldDescriptor &srcField, const ROOT::RFieldDescriptor &dstField,
                                       RDescriptorsComparison &result, std::vector<std::string> &errors)
{
   const auto &srcColumns = srcField.GetLogicalColumnIds();
   const auto &dstColumns = dstField.GetLogicalColumnIds();
 
   // Fields must have the same cardinality
   const std::uint32_t srcColCardinality = srcField.GetColumnCardinality();
   const std::uint32_t dstColCardinality = dstField.GetColumnCardinality();
   if (srcColCardinality != dstColCardinality) {
      std::stringstream ss;
      ss << "Field `" << srcField.GetFieldName()
         << "` has a different column cardinality than previously-seen field with the same name (old: "
         << dstColCardinality << ", new: " << srcColCardinality << ")";
      errors.push_back(ss.str());
      return;
   }
 
   if (srcColCardinality == 0)
      return; // no columns to match
 
   const auto srcNColReprs = srcColumns.size() / srcColCardinality;
   const auto dstNColReprs = dstColumns.size() / dstColCardinality;
   std::uint32_t nextDstReprIndex = dstNColReprs;
 
   // For each column representation of the source, check if it matches one in the descriptor.
   // If so, and if it doesn't match the destination's repr index, add a mapping for it.
   // If nothing matches, schedule the column representation to be added later.
   // NOTE: this has quadratic complexity but the numbers involved are small so it's fine.
   for (auto srcReprIdx = 0u; srcReprIdx < srcNColReprs; ++srcReprIdx) {
      std::int64_t matchingRepr = -1;
      for (auto dstReprIdx = 0u; dstReprIdx < dstNColReprs; ++dstReprIdx) {
         bool matches = true;
         for (auto reprColIdx = 0u; reprColIdx < srcColCardinality; ++reprColIdx) {
            const auto srcColId = srcColumns[srcReprIdx * srcColCardinality + reprColIdx];
            const auto &srcCol = srcDesc.GetColumnDescriptor(srcColId);
            const auto dstColId = dstColumns[dstReprIdx * dstColCardinality + reprColIdx];
            const auto &dstCol = dstDesc.GetColumnDescriptor(dstColId);
            if (srcCol.GetType() != dstCol.GetType()) {
               matches = false;
               break;
            }
         }
 
         if (matches) {
            // If this column representation matches by column type, we need to make sure that it also has
            // matching column metadata. Since we currently do not support multiple column representations
            // that only differ by such metadata, we forbid merging such columns (e.g. we cannot merge two
            // Real32Trunc columns with different bit widths). This could technically be supported, but it
            // would require significant effort, so we currently don't.
            for (auto reprColIdx = 0u; reprColIdx < srcColCardinality; ++reprColIdx) {
               const auto srcColId = srcColumns[srcReprIdx * srcColCardinality + reprColIdx];
               const auto &srcCol = srcDesc.GetColumnDescriptor(srcColId);
               const auto dstColId = dstColumns[dstReprIdx * dstColCardinality + reprColIdx];
               const auto &dstCol = dstDesc.GetColumnDescriptor(dstColId);
               if (srcCol.GetType() != dstCol.GetType() || srcCol.GetBitsOnStorage() != dstCol.GetBitsOnStorage() ||
                   srcCol.GetValueRange() != dstCol.GetValueRange()) {
                  matches = false;
                  break;
               }
            }
 
            if (matches) {
               // We found a valid matching representation.
               matchingRepr = dstReprIdx;
               break;
            }
         }
      }
 
      if (errors.empty()) {
         if (matchingRepr >= 0 && matchingRepr != srcReprIdx) {
            // a different matching representation was found
            assert(matchingRepr < std::numeric_limits<std::uint32_t>::max());
            result.fColReprMappings[&dstField].push_back(
               RColReprMapping{srcReprIdx, static_cast<std::uint32_t>(matchingRepr)});
         } else if (matchingRepr < 0) {
            // this representation was not found in the destination: add it
            std::vector<ROOT::Internal::RColumnFormat> newRepr;
            newRepr.reserve(srcColCardinality);
            std::uint32_t firstElemIdx = 0;
            for (auto reprColIdx = 0u; reprColIdx < srcColCardinality; ++reprColIdx) {
               const auto srcColId = srcColumns[srcReprIdx * srcColCardinality + reprColIdx];
               const auto &srcCol = srcDesc.GetColumnDescriptor(srcColId);
               // All added columns are supposed to have the same firstElementIndex
               assert(firstElemIdx == 0 || firstElemIdx == srcCol.GetFirstElementIndex());
               firstElemIdx = srcCol.GetFirstElementIndex();
               auto &reprElement = newRepr.emplace_back();
               reprElement.fType = srcCol.GetType();
               reprElement.fBitWidth = srcCol.GetBitsOnStorage();
               reprElement.fValueRange = srcCol.GetValueRange();
            }
            RColReprExtension extension{{srcReprIdx, nextDstReprIndex}, newRepr, firstElemIdx};
            nextDstReprIndex += newRepr.size();
            result.fColReprExtensions[&dstField].push_back(extension);
            result.fColReprMappings[&dstField].push_back(extension);
         }
      }
   }
}
 
static ROOT::RResult<RDescriptorsComparison> FillCommonAndExtraFields(const ROOT::RNTupleDescriptor &dst,
                                                                      const ROOT::RNTupleDescriptor &src,
                                                                      bool forAttributeMerging)
{
   RDescriptorsComparison res;
 
   // collect common and extra dst fields
   for (const auto &dstField : dst.GetTopLevelFields()) {
      const auto srcFieldId = src.FindFieldId(dstField.GetFieldName());
      if (srcFieldId != ROOT::kInvalidDescriptorId) {
         const auto &srcField = src.GetFieldDescriptor(srcFieldId);
         res.fCommonFields.push_back({srcField, dstField});
      } else {
         res.fExtraDstFields.emplace_back(&dstField);
      }
   }
 
   // collect extra src fields
   for (const auto &srcField : src.GetTopLevelFields()) {
      const auto dstFieldId = dst.FindFieldId(srcField.GetFieldName());
      if (dstFieldId == ROOT::kInvalidDescriptorId) {
         res.fExtraSrcFields.push_back(&srcField);
      }
   }
 
   if (forAttributeMerging) {
      using namespace ROOT::Internal::RNTupleAttributes;
      const auto userDataFieldId = src.FindFieldId(kMetaFieldNames[kMetaFieldIndex_UserData]);
      if (userDataFieldId == ROOT::kInvalidDescriptorId) {
         return R__FAIL("attribute user data field not found. The RNTuple is corrupted and cannot be merged.");
      }
      const auto &userDataField = src.GetFieldDescriptor(userDataFieldId);
      for (const auto subfieldId : userDataField.GetLinkIds()) {
         const auto subfieldName = src.GetQualifiedFieldName(subfieldId);
         const auto dstFieldId = dst.FindFieldId(subfieldName);
         if (dstFieldId == ROOT::kInvalidDescriptorId) {
            const auto &subfield = src.GetFieldDescriptor(subfieldId);
            res.fExtraSrcSubfields[userDataFieldId].push_back(&subfield);
         }
      }
   }
 
   return res;
}
 
static void VerifyProjectedFieldsCompatibility(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTupleDescriptor &src,
                                               const ROOT::RFieldDescriptor &dstField,
                                               const ROOT::RFieldDescriptor &srcField, std::vector<std::string> &errors)
{
   bool projCompatible = srcField.IsProjectedField() == dstField.IsProjectedField();
   if (!projCompatible) {
      std::stringstream ss;
      ss << "Field `" << dstField.GetFieldName()
         << "` is incompatible with previously-seen field with that name because the "
         << (srcField.IsProjectedField() ? "new" : "old") << " one is projected and the other isn't";
      errors.push_back(ss.str());
   } else if (srcField.IsProjectedField()) {
      // if both fields are projected, verify that they point to the same real field
      const auto srcName = src.GetQualifiedFieldName(srcField.GetProjectionSourceId());
      const auto dstName = dst.GetQualifiedFieldName(dstField.GetProjectionSourceId());
      if (srcName != dstName) {
         std::stringstream ss;
         ss << "Field `" << dstField.GetFieldName()
            << "` is projected to a different field than a previously-seen field with the same name (old: " << dstName
            << ", new: " << srcName << ")";
         errors.push_back(ss.str());
      }
   }
}
 
/// Compares the top level fields of `dst` and `src` and determines whether they can be merged or not.
/// In addition, returns the differences between `dst` and `src`'s structures
static ROOT::RResult<RDescriptorsComparison> CompareDescriptorStructure(const ROOT::RNTupleDescriptor &dst,
                                                                        const ROOT::RNTupleDescriptor &src,
                                                                        bool forAttributeMerging = false)
{
   // Cases:
   // 1. dst == src
   // 2. dst has fields that src hasn't
   // 3. src has fields that dst hasn't
   // 4. dst and src have fields that differ (compatible or incompatible)
 
   auto descCmpRes = FillCommonAndExtraFields(dst, src, forAttributeMerging);
   if (!descCmpRes)
      return R__FORWARD_RESULT(descCmpRes);
 
   auto res = descCmpRes.Unwrap();
   std::vector<std::string> slowMergeErrors;
   std::vector<std::string> errors;
 
   // Check compatibility of common fields
   auto fieldsToCheck = res.fCommonFields;
   const auto nTopLevelFields = fieldsToCheck.size();
 
   // NOTE: using index-based for loop because the collection may get extended by the iteration
   for (std::size_t fieldIdx = 0; fieldIdx < fieldsToCheck.size(); ++fieldIdx) {
      const auto &field = fieldsToCheck[fieldIdx];
 
      // NOTE: field.fSrc and field.fDst have the same name by construction
      const auto &fieldName = field.fSrc->GetFieldName();
 
      // Require that fields are both projected or both not projected
      auto &errs = (fieldIdx < nTopLevelFields) ? slowMergeErrors : errors;
      VerifyProjectedFieldsCompatibility(dst, src, *field.fDst, *field.fSrc, errs);
 
      // Require that fields types match
      // TODO(gparolini): allow non-identical but compatible types
      const auto &srcTyName = ROOT::Internal::GetRenormalizedTypeName(field.fSrc->GetTypeName());
      // This is already renormalized by construction (see RNTupleDescriptorBuilder::SetSchemaFromExisting)
      const auto &dstTyName = field.fDst->GetTypeName();
      if (srcTyName != dstTyName) {
         std::stringstream ss;
         ss << "Field `" << fieldName
            << "` has a type incompatible with a previously-seen field with the same name: (old: " << dstTyName
            << ", new: " << srcTyName << ")";
         errors.push_back(ss.str());
      }
 
      // Require that type checksums match
      const auto srcTyChk = field.fSrc->GetTypeChecksum();
      const auto dstTyChk = field.fDst->GetTypeChecksum();
      if (srcTyChk && dstTyChk && *srcTyChk != *dstTyChk) {
         std::stringstream ss;
         ss << "Field `" << field.fSrc->GetFieldName()
            << "` has a different type checksum than previously-seen field with the same name";
         errors.push_back(ss.str());
      }
 
      // Require that type versions match
      const auto srcTyVer = field.fSrc->GetTypeVersion();
      const auto dstTyVer = field.fDst->GetTypeVersion();
      if (srcTyVer != dstTyVer) {
         std::stringstream ss;
         ss << "Field `" << field.fSrc->GetFieldName()
            << "` has a different type version than previously-seen field with the same name (old: " << dstTyVer
            << ", new: " << srcTyVer << ")";
         errors.push_back(ss.str());
      }
 
      // Require that field versions match
      const auto srcFldVer = field.fSrc->GetFieldVersion();
      const auto dstFldVer = field.fDst->GetFieldVersion();
      if (srcFldVer != dstFldVer) {
         std::stringstream ss;
         ss << "Field `" << field.fSrc->GetFieldName()
            << "` has a different field version than previously-seen field with the same name (old: " << dstFldVer
            << ", new: " << srcFldVer << ")";
         errors.push_back(ss.str());
      }
 
      const auto srcRole = field.fSrc->GetStructure();
      const auto dstRole = field.fDst->GetStructure();
      if (srcRole != dstRole) {
         std::stringstream ss;
         ss << "Field `" << field.fSrc->GetFieldName()
            << "` has a different structural role than previously-seen field with the same name (old: " << dstRole
            << ", new: " << srcRole << ")";
         errors.push_back(ss.str());
      }
 
      // Require that column representations match
      if (!field.fSrc->IsProjectedField()) {
         MatchColumnRepresentations(src, dst, *field.fSrc, *field.fDst, res, errors);
      }
 
      // Require that subfields are compatible
      const auto &srcLinks = field.fSrc->GetLinkIds();
      const auto &dstLinks = field.fDst->GetLinkIds();
      if (srcLinks.size() != dstLinks.size()) {
         std::stringstream ss;
         ss << "Field `" << field.fSrc->GetFieldName()
            << "` has a different number of children than previously-seen field with the same name (old: "
            << dstLinks.size() << ", new: " << srcLinks.size() << ")";
         errors.push_back(ss.str());
      } else {
         for (std::size_t linkIdx = 0, linkNum = srcLinks.size(); linkIdx < linkNum; ++linkIdx) {
            const auto &srcSubfield = src.GetFieldDescriptor(srcLinks[linkIdx]);
            const auto &dstSubfield = dst.GetFieldDescriptor(dstLinks[linkIdx]);
            fieldsToCheck.push_back(RCommonField{srcSubfield, dstSubfield});
         }
      }
   }
 
   std::string errMsg;
   if (!slowMergeErrors.empty()) {
      // If there are slow merge errors, the source cannot be merged at all.
      for (const auto &err : slowMergeErrors)
         errMsg += std::string("\n  * ") + err;
 
      errMsg = errMsg.substr(1); // strip initial newline
      return R__FAIL(errMsg);
   }
 
   // Source can only be slow merged.
   if (!errors.empty()) {
      for (const auto &err : errors)
         errMsg += std::string("\n  * ") + err;
      errMsg = errMsg.substr(1); // strip initial newline
      res.fFastMergeErrors = errMsg;
   }
 
   return ROOT::RResult(res);
}
 
// Restores the truncation/quantization parameters (bit width, value range) of a field recreated
// from an on-disk descriptor, e.g. by ExtendDestinationModel(). RFieldDescriptor::CreateField only
// rebuilds the type structure; the parameters live in the in-memory field (RRealField::fBitWidth,
// fValueMin, fValueMax), so pinning the column representative alone leaves stale defaults behind,
// which are out of range for kReal32Trunc. Recurses into subfields, matched by on-disk ID, so that
// truncated/quantized members of late records and collections are restored as well. Nodes whose
// source column is not truncated/quantized, or which span several representations, are left alone.
static ROOT::RResult<void> RestoreRealFieldParams(ROOT::RFieldBase &field, const ROOT::RFieldDescriptor &fieldDesc,
                                                  const ROOT::RNTupleDescriptor &srcDesc)
{
   const auto &logicalIds = fieldDesc.GetLogicalColumnIds();
   if (logicalIds.size() == 1) {
      const auto &column = srcDesc.GetColumnDescriptor(logicalIds[0]);
      const auto type = column.GetType();
      if (type == ROOT::ENTupleColumnType::kReal32Trunc || type == ROOT::ENTupleColumnType::kReal32Quant) {
         auto *realFloat = dynamic_cast<ROOT::RRealField<float> *>(&field);
         auto *realDouble = dynamic_cast<ROOT::RRealField<double> *>(&field);
         if (!realFloat && !realDouble) {
            return R__FAIL("field `" + field.GetQualifiedFieldName() +
                           "` has a Real32Trunc/Real32Quant column but is not an RRealField");
         }
         const auto bits = column.GetBitsOnStorage();
         try {
            if (type == ROOT::ENTupleColumnType::kReal32Trunc) {
               if (realFloat)
                  realFloat->SetTruncated(bits);
               else
                  realDouble->SetTruncated(bits);
            } else {
               const auto valueRange = column.GetValueRange();
               if (!valueRange) {
                  return R__FAIL("field `" + field.GetQualifiedFieldName() +
                                 "` has a Real32Quant column with no value range");
               }
               if (realFloat) {
                  realFloat->SetQuantized(bits,
                                          {static_cast<float>(valueRange->fMin), static_cast<float>(valueRange->fMax)});
               } else {
                  realDouble->SetQuantized(bits, {valueRange->fMin, valueRange->fMax});
               }
            }
         } catch (const ROOT::RException &ex) {
            return R__FAIL(ex.what());
         }
      }
   }
   for (auto *subfield : field.GetMutableSubfields()) {
      // CreateField() may materialize in-memory class members that the source descriptor does not
      // have (e.g. members added to the class after the file was written); those keep an invalid
      // on-disk ID and carry no source column, so there is nothing to restore for them.
      if (subfield->GetOnDiskId() == ROOT::kInvalidDescriptorId)
         continue;
      auto restored = RestoreRealFieldParams(*subfield, srcDesc.GetFieldDescriptor(subfield->GetOnDiskId()), srcDesc);
      if (!restored)
         return R__FORWARD_ERROR(restored);
   }
   return ROOT::RResult<void>::Success();
}
 
// Applies late model extension to `mergeData.fDestination`, adding all `descCmp.fExtraSrcFields` to it.
[[nodiscard]]
static ROOT::RResult<void>
ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstModel,
                       ROOT::Internal::RPageSink &destination, const ROOT::RNTupleDescriptor &srcDesc,
                       ROOT::NTupleSize_t nEntries, bool verbose, bool forSlowMerge)
{
   const auto &newFields = descCmp.fExtraSrcFields;
 
   dstModel.Unfreeze();
   ROOT::Internal::RNTupleModelChangeset changeset{dstModel};
 
   if (verbose) {
      std::string msg = "destination doesn't contain field";
      if (newFields.size() > 1)
         msg += 's';
      msg += ' ';
      msg += std::accumulate(newFields.begin(), newFields.end(), std::string{}, [](const auto &acc, const auto *field) {
         return acc + (acc.length() ? ", " : "") + '`' + field->GetFieldName() + '`';
      });
      R__LOG_INFO(NTupleMergeLog()) << msg << ": adding " << (newFields.size() > 1 ? "them" : "it")
                                    << " to the destination model (entry #" << nEntries << ").";
   }
 
   ROOT::RCreateFieldOptions cfOpts;
   cfOpts.SetEmulateUnknownTypes(true);
 
   changeset.fAddedFields.reserve(newFields.size());
   // First add all non-projected fields...
   for (const auto *fieldDesc : newFields) {
      if (fieldDesc->IsProjectedField())
         continue;
 
      auto field = fieldDesc->CreateField(srcDesc, cfOpts);
      if (!forSlowMerge) {
         // Explicitly set the field representatives. This prevents UpdateSchema() from changing our column
         // representations via AutoAdjustColumnTypes.
         ROOT::RFieldBase::ColumnRepresentation_t representatives;
         for (const auto &colId : fieldDesc->GetLogicalColumnIds()) {
            const auto &column = srcDesc.GetColumnDescriptor(colId);
            representatives.push_back(column.GetType());
         }
         field->SetColumnRepresentatives({representatives});
         // Pinning the representative is not enough for truncated/quantized columns: their parameters
         // are stored in the in-memory field, which CreateField() leaves at defaults.
         auto restored = RestoreRealFieldParams(*field, *fieldDesc, srcDesc);
         if (!restored)
            return R__FORWARD_ERROR(restored);
      }
      changeset.AddField(std::move(field));
   }
   // ...then add all projected fields...
   for (const auto *fieldDesc : newFields) {
      if (!fieldDesc->IsProjectedField())
         continue;
 
      ROOT::Internal::RProjectedFields::FieldMap_t fieldMap;
      auto field = fieldDesc->CreateField(srcDesc, cfOpts);
      const auto sourceId = fieldDesc->GetProjectionSourceId();
      const auto &sourceField = dstModel.GetConstField(srcDesc.GetQualifiedFieldName(sourceId));
      fieldMap[field.get()] = &sourceField;
 
      for (const auto &subfield : *field) {
         const auto &subFieldDesc = srcDesc.GetFieldDescriptor(subfield.GetOnDiskId());
         const auto subSourceId = subFieldDesc.GetProjectionSourceId();
         const auto &subSourceField = dstModel.GetConstField(srcDesc.GetQualifiedFieldName(subSourceId));
         fieldMap[&subfield] = &subSourceField;
      }
      changeset.fAddedProjectedFields.emplace_back(field.get());
      ROOT::Internal::GetProjectedFieldsOfModel(dstModel).Add(std::move(field), fieldMap);
   }
   // ...finally add all subfields.
   if (forSlowMerge) { // TODO(gparolini): figure out if this makes sense also in default merging
      for (const auto &[parentId, subfields] : descCmp.fExtraSrcSubfields) {
         const auto parentFQName = srcDesc.GetQualifiedFieldName(parentId);
         // TODO(gparolini): should we RegisterSubfield(parentFQName) here if the parent is not a top-level field?
         for (const auto *subfield : subfields) {
            auto field = subfield->CreateField(srcDesc, cfOpts);
            changeset.AddField(std::move(field), parentFQName);
         }
      }
   }
   dstModel.Freeze();
   try {
      // FIXME: here we are connecting the new fields/columns to the sink!
      // We should avoid doing that, as all other non-extended fields never get connected (and we don't
      // need to connect these either in principle).
      // NOTE: this calls AutoAdjustColumnTypes, but we have set the column representations of all fields
      // explicitly, so it will not change it under the hood.
      destination.UpdateSchema(changeset, nEntries);
   } catch (const ROOT::RException &ex) {
      return R__FAIL(ex.what());
   }
 
   auto &commonFields = descCmp.fCommonFields;
   commonFields.reserve(commonFields.size() + newFields.size());
   // NOTE(gparolini): Insert the new fields at the beginning of `commonFields`.
   // We need to make sure the extended fields appear before all other common fields for the following reason:
   // in general, when we GatherColumnInfos we (potentially) assign new column output ids in field order; this
   // assignment happens whenever we find new columns, which happens in 3 cases:
   //   1. we are in the first source and we're adding the first set of (common) fields;
   //   2. we are adding a new set of extended common fields (this is done in this function);
   //   3. we are adding new column representations for fields that we already had before processing this source.
   //
   // It's important that the output id assigned to the new columns is coherent with the order of the column descriptors
   // as they appear in the header and footer of the destination RNTuple.
   // This is in turn determined by the order by which we append new columns to the dst descriptor during the merging
   // process.
   //
   // Now let's consider the three cases listed above.
   // Ignoring the trivial case (1), the order of operations for each source is:
   //   - call ExtendDestinationModel (case 2)
   //       (this adds both new fields and column descriptors; see the UpdateSchema call above)
   //   - add new column representations (case 3)
   //       (this only adds column descriptors, see the call to AddColumnRepresentation)
   //
   // Since we call ExtendDestinationModel (this function) *before* adding the new column representations,
   // the dst descriptor always gets updated with the new column descriptors coming from the extended fields before
   // it gets updated with the extended column representations.
   //
   // However, in GatherColumnInfos, the new column output ids are added sequentially in *field* order and the fields
   // containing the new column representations are already in that list from earlier! So, to make sure the new output
   // ids are assigned to our extended fields first, we push them in from on the list so they are visited first.
   const auto &dstDesc = destination.GetDescriptor();
   for (auto it = newFields.rbegin(); it != newFields.rend(); ++it) {
      const auto *field = *it;
      const auto newFieldInDstId = dstDesc.FindFieldId(field->GetFieldName());
      const auto &newFieldInDst = dstDesc.GetFieldDescriptor(newFieldInDstId);
      commonFields.insert(commonFields.begin(), RCommonField{*field, newFieldInDst});
   }
 
   return ROOT::RResult<void>::Success();
}
 
static std::optional<std::type_index> ColumnInMemoryType(std::string_view fieldType, ENTupleColumnType onDiskType)
{
   if (onDiskType == ENTupleColumnType::kIndex32 || onDiskType == ENTupleColumnType::kSplitIndex32 ||
       onDiskType == ENTupleColumnType::kIndex64 || onDiskType == ENTupleColumnType::kSplitIndex64)
      return typeid(ROOT::Internal::RColumnIndex);
 
   if (onDiskType == ENTupleColumnType::kSwitch)
      return typeid(ROOT::Internal::RColumnSwitch);
 
   // clang-format off
   if (fieldType == "bool")          return typeid(bool);
   if (fieldType == "std::byte")     return typeid(std::byte);
   if (fieldType == "char")          return typeid(char);
   if (fieldType == "std::int8_t")   return typeid(std::int8_t);
   if (fieldType == "std::uint8_t")  return typeid(std::uint8_t);
   if (fieldType == "std::int16_t")  return typeid(std::int16_t);
   if (fieldType == "std::uint16_t") return typeid(std::uint16_t);
   if (fieldType == "std::int32_t")  return typeid(std::int32_t);
   if (fieldType == "std::uint32_t") return typeid(std::uint32_t);
   if (fieldType == "std::int64_t")  return typeid(std::int64_t);
   if (fieldType == "std::uint64_t") return typeid(std::uint64_t);
   if (fieldType == "float")         return typeid(float);
   if (fieldType == "double")        return typeid(double);
   // clang-format on
 
   // if the type is not one of those above, we use the default in-memory type.
   return std::nullopt;
}
 
static void PrefillColumnMap(const ROOT::RNTupleDescriptor &desc, const ROOT::RFieldDescriptor &fieldDesc,
                             ColumnIdMap_t &colIdMap, const std::string &prefix = "")
{
   std::string name = prefix + '.' + fieldDesc.GetFieldName();
   for (const auto &colId : fieldDesc.GetLogicalColumnIds()) {
      const auto &colDesc = desc.GetColumnDescriptor(colId);
      RColumnOutInfo info{};
      info.fColumnId = colDesc.GetLogicalId();
      const auto colName =
         name + '.' + std::to_string(colDesc.GetIndex()) + '.' + std::to_string(colDesc.GetRepresentationIndex());
      colIdMap[colName] = info;
   }
 
   for (const auto &subId : fieldDesc.GetLinkIds()) {
      const auto &subfield = desc.GetFieldDescriptor(subId);
      PrefillColumnMap(desc, subfield, colIdMap, name);
   }
}
 
static void AddColumnExtensionsInFieldOrder(
   const ROOT::RFieldDescriptor &field, const ROOT::RNTupleDescriptor &desc,
   const FieldCollectionMap_t<RColReprExtension> &extensions,
   std::vector<std::pair<const ROOT::RFieldDescriptor *, std::vector<RColReprExtension>>> &outExtensions,
   std::unordered_map<ROOT::DescriptorId_t, std::vector<const ROOT::RFieldDescriptor *>> &outProjectionPointees)
{
   const auto it = extensions.find(&field);
   if (it != extensions.end())
      outExtensions.emplace_back(it->first, it->second);
 
   if (field.IsProjectedField())
      outProjectionPointees[field.GetProjectionSourceId()].push_back(&field);
 
   for (auto childId : field.GetLinkIds()) {
      const auto &child = desc.GetFieldDescriptor(childId);
      AddColumnExtensionsInFieldOrder(child, desc, extensions, outExtensions, outProjectionPointees);
   }
}
 
namespace {
// Collection of common methods and data to be used for any kind of RNTuples merging.
class RNTupleMergeStrategy {
protected:
   // Fixed during the whole merging. Note that these references are owned by the RNTupleMerger and share its lifetime.
   std::unique_ptr<ROOT::RNTupleModel> fDstModel;
   ROOT::Internal::RPagePersistentSink &fDestination;
   const ROOT::RNTupleDescriptor &fDstDescriptor;
   RNTupleMergeOptions fMergeOpts;
 
   // Changes at every source. InitSource() is supposed to update this right after Attaching the source.
   const ROOT::RNTupleDescriptor *fSrcDescriptor = nullptr;
 
public:
   RNTupleMergeStrategy(std::unique_ptr<RNTupleModel> dstModel, ROOT::Internal::RPagePersistentSink &dst,
                        const RNTupleMergeOptions &mergeOpts)
      : fDstModel(std::move(dstModel)), fDestination(dst), fDstDescriptor(dst.GetDescriptor()), fMergeOpts(mergeOpts)
   {
   }
 
   virtual ~RNTupleMergeStrategy() = default;
 
   // Returns the number of entries currently written to the destination.
   // This is only accurate before or after a call to MergeSource(), not during it.
   virtual ROOT::NTupleSize_t GetNDstEntries() const = 0;
 
   // Invoked as the first thing when we start processing a new source
   virtual void InitSource(RPageSource &source) = 0;
 
   // Invoked when the destination model must be late model extended (only invoked in case of union merging)
   [[nodiscard]] virtual ROOT::RResult<void> ExtendDestinationModel(RDescriptorsComparison &descCmp) = 0;
 
   // Invoked when merging one source
   [[nodiscard]] virtual ROOT::RResult<void> MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) = 0;
 
   // Invoked at the end of the whole process, just before committing the dataset
   virtual void BeforeCommit() = 0;
};
 
/// Merge Actions used internally to merge attributes.
class RNTupleAttrMergeActions final : public RNTupleMergeActions {
   /// How much do we need to shift the _rangeStart field by. This changes on each new source (and is equal to the
   /// total number of entries committed at the end of the previous source).
   ROOT::NTupleSize_t fAttrEntryStartOffset = 0;
   const RNTupleMergeStrategy &fParentStrategy;
   std::shared_ptr<ROOT::NTupleSize_t> fRangeStartPtr;
 
public:
   RNTupleAttrMergeActions(const RNTupleMergeStrategy &parentStrategy, ROOT::NTupleSize_t initialStartOffset)
      : fAttrEntryStartOffset(initialStartOffset), fParentStrategy(parentStrategy)
   {
   }
 
   ENTupleMergeActionResult OnFill(ROOT::REntry &entry, const RNTupleMergeActionContext &) final
   {
      using namespace ROOT::Internal::RNTupleAttributes;
 
      // Cache pointer per source
      if (!fRangeStartPtr)
         fRangeStartPtr = entry.GetPtr<ROOT::NTupleSize_t>(kMetaFieldNames[kMetaFieldIndex_RangeStart]);
 
      // Shift _rangeStart field by the number of main entries we already written in the destination
      *fRangeStartPtr += fAttrEntryStartOffset;
      return ENTupleMergeActionResult::kEmit;
   }
 
   ENTupleMergeActionResult OnSourceEnd(ROOT::REntry &, const RNTupleMergeActionContext &) final
   {
      fAttrEntryStartOffset = fParentStrategy.GetNDstEntries();
      fRangeStartPtr = nullptr; // invalidate cached ptr every time we switch source
      return ENTupleMergeActionResult::kDrop;
   }
};
 
// =========================================================================================
//
// MergeStrategyDefault
//
// =========================================================================================
// Strategy used for L1, L2 and L3 merging
// (https://github.com/root-project/root/blob/master/tree/ntuple/doc/Merging.md#glossary).
class RNTupleMergeStrategyDefault final : public RNTupleMergeStrategy {
   std::unique_ptr<ROOT::Internal::RPageAllocator> fPageAlloc = std::make_unique<ROOT::Internal::RPageAllocatorHeap>();
   std::optional<ROOT::Experimental::TTaskGroup> fTaskGroup;
   std::unordered_map<ROOT::DescriptorId_t, std::vector<const ROOT::RFieldDescriptor *>> fProjectionPointees;
 
   std::vector<RColumnMergeInfo> fColumns;
   // Maps input column IDs to output IDs
   ColumnIdMap_t fColumnIdMap;
 
   [[nodiscard]] ROOT::RResult<void>
   MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool, const ROOT::RClusterDescriptor &clusterDesc,
                      std::span<RColumnMergeInfo> commonColumns, const RCluster::ColumnSet_t &commonColumnSet,
                      RSealedPageMergeData &sealedPageData);
   [[nodiscard]]
   ROOT::RResult<void> GenerateZeroPagesForColumns(size_t nEntriesToGenerate, std::span<const RColumnMergeInfo> columns,
                                                   RSealedPageMergeData &sealedPageData);
 
   void AddColumnsFromField(std::vector<RColumnMergeInfo> &columns, const ROOT::RNTupleDescriptor &srcDesc,
                            const FieldCollectionMap_t<RColReprMapping> &colReprMappings,
                            const ROOT::RFieldDescriptor &srcFieldDesc, const ROOT::RFieldDescriptor &dstFieldDesc,
                            const std::string &prefix = "");
 
   RColumnInfoGroup GatherColumnInfos(const RDescriptorsComparison &descCmp, const ROOT::RNTupleDescriptor &srcDesc);
 
   [[nodiscard]] ROOT::RResult<void> MergeSourceClusters(RPageSource &source, std::span<RColumnMergeInfo> commonColumns,
                                                         std::span<const RColumnMergeInfo> extraDstColumns);
 
public:
   RNTupleMergeStrategyDefault(std::unique_ptr<RNTupleModel> dstModel, ROOT::Internal::RPagePersistentSink &dst,
                               const RNTupleMergeOptions &mergeOpts)
      : RNTupleMergeStrategy(std::move(dstModel), dst, mergeOpts)
   {
#ifdef R__USE_IMT
      if (ROOT::IsImplicitMTEnabled())
         fTaskGroup = ROOT::Experimental::TTaskGroup();
#endif
 
      if (fDstModel) {
         // If this is an incremental merging, pre-fill the column id map with the existing destination ids.
         // Otherwise we would generate new output ids that may not match the ones in the destination!
         for (const auto &field : dst.GetDescriptor().GetTopLevelFields()) {
            PrefillColumnMap(dst.GetDescriptor(), field, fColumnIdMap);
         }
      }
   }
 
   ROOT::NTupleSize_t GetNDstEntries() const final
   {
      // Since with default merging we always commit clusters at source boundary, we can simply rely on
      // the sink's internal counter.
      return fDestination.GetNEntries();
   }
 
   void InitSource(RPageSource &source) final
   {
      source.Attach(RNTupleSerializer::EDescriptorDeserializeMode::kForWriting);
      source.LoadAllPageLists();
      fSrcDescriptor = &source.GetSharedDescriptorGuard().GetConstRef();
 
      if (!fDstModel) {
         constexpr bool copyClusters = false;
         fDstModel = fDestination.InitFromDescriptor(*fSrcDescriptor, copyClusters);
      }
   }
 
   [[nodiscard]] ROOT::RResult<void> ExtendDestinationModel(RDescriptorsComparison &descCmp) final
   {
      // late model extension for all fExtraSrcFields in Union mode
      auto res = ::ExtendDestinationModel(descCmp, *fDstModel, fDestination, *fSrcDescriptor, GetNDstEntries(),
                                          fMergeOpts.fExtraVerbose,
                                          /*forSlowMerge=*/false);
      return res;
   }
 
   void BeforeCommit() final {}
 
   [[nodiscard]] ROOT::RResult<void> MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) final;
};
} // namespace
 
// Given a field, fill `columns` and `mergeData.fColumnIdMap` with information about all columns belonging to it and
// its subfields. `fColumnIdMap` is used to map matching columns from different sources to the same output
// column in the destination. We match columns by their "fully qualified name", which is the concatenation of their
// ancestor fields' names and the column index. By this point, since we called `CompareDescriptorStructure()`
// earlier, we should be guaranteed that two matching columns will have at least compatible representations.
// This function is recursive as it needs to call itself on the entire subfield hierarchy of the source field.
// NOTE: srcFieldDesc and dstFieldDesc may alias.
void RNTupleMergeStrategyDefault::AddColumnsFromField(std::vector<RColumnMergeInfo> &columns,
                                                      const ROOT::RNTupleDescriptor &srcDesc,
                                                      const FieldCollectionMap_t<RColReprMapping> &colReprMappings,
                                                      const ROOT::RFieldDescriptor &srcFieldDesc,
                                                      const ROOT::RFieldDescriptor &dstFieldDesc,
                                                      const std::string &prefix)
{
   std::string name = prefix + '.' + srcFieldDesc.GetFieldName();
 
   // We don't want to try and merge alias columns. Note that subfields of projected fields
   // must also be projected, so we don't need to check them.
   if (srcFieldDesc.IsProjectedField())
      return;
 
   const auto &columnIds = srcFieldDesc.GetLogicalColumnIds();
   columns.reserve(columns.size() + columnIds.size());
 
   for (auto i = 0u; i < srcFieldDesc.GetLogicalColumnIds().size(); ++i) {
      auto srcColumnId = srcFieldDesc.GetLogicalColumnIds()[i];
      const auto &srcColumn = srcDesc.GetColumnDescriptor(srcColumnId);
 
      RColumnMergeInfo info{};
      info.fInputId = srcColumn.GetPhysicalId();
      // NOTE(gparolini): the parent field is used when synthesizing zero pages, which happens in 2 situations:
      // 1. when adding extra dst columns (in which case we need to synthesize zero pages for the incoming src), and
      // 2. when merging a deferred column into an existing column (in which case we need to fill the "hole" with
      // zeroes). For the first case srcFieldDesc and dstFieldDesc are the same (see the calling site of this
      // function), but for the second case they're not, and we need to pick the source field because we will then
      // check the column's *input* id inside fParentFieldDescriptor to see if it's a suppressed column (see
      // GenerateZeroPagesForColumns()).
      info.fParentFieldDescriptor = &srcFieldDesc;
      // Save the parent field descriptor since this may be either the source or destination descriptor depending on
      // whether this is an extraDstField or a commonField. We will need this in GenerateZeroPagesForColumns() to
      // properly walk up the field hierarchy.
      info.fParentNTupleDescriptor = &srcDesc;
 
      const auto mappingsIt = colReprMappings.find(&dstFieldDesc);
      std::uint16_t reprIndex = srcColumn.GetRepresentationIndex();
      if (mappingsIt != colReprMappings.end()) {
         if (auto outReprIdx = FindColumnReprMapping(mappingsIt->second, reprIndex); outReprIdx)
            reprIndex = *outReprIdx;
      }
 
      info.fColumnName = name + '.' + std::to_string(srcColumn.GetIndex()) + '.' + std::to_string(reprIndex);
 
      ENTupleColumnType columnType = ENTupleColumnType::kUnknown;
 
      if (auto it = fColumnIdMap.find(info.fColumnName); it != fColumnIdMap.end()) {
         // We had already added this column to the column id map: just copy its data.
         info.fOutputId = it->second.fColumnId;
         info.fOutputReprIndex = reprIndex;
      } else {
         // New column: assign it the next ouput id.
         info.fOutputId = fColumnIdMap.size();
         // NOTE(gparolini): map the representation index of src column to that of dst column.
         // This mapping is only relevant for common columns and it's done to ensure we have the correct representation
         // index in the output column metadata.
         assert(dstFieldDesc.GetColumnCardinality() == srcFieldDesc.GetColumnCardinality());
         const auto dstColumnIndex = reprIndex * dstFieldDesc.GetColumnCardinality() + srcColumn.GetIndex();
         const auto dstColumnId = dstFieldDesc.GetLogicalColumnIds()[dstColumnIndex];
         const auto &dstColumn = fDstDescriptor.GetColumnDescriptor(dstColumnId);
         columnType = dstColumn.GetType();
         info.fOutputReprIndex = reprIndex;
         fColumnIdMap[info.fColumnName] = RColumnOutInfo{info.fOutputId};
      }
 
      if (fMergeOpts.fExtraVerbose) {
         R__LOG_INFO(NTupleMergeLog()) << "Adding column " << info.fColumnName << " with log.id " << srcColumnId
                                       << ", phys.id " << srcColumn.GetPhysicalId() << ", type "
                                       << RColumnElementBase::GetColumnTypeName(srcColumn.GetType()) << " -> log.id "
                                       << info.fOutputId << ", type "
                                       << RColumnElementBase::GetColumnTypeName(columnType);
      }
 
      // Since we disallow merging fields of different types, src and dstFieldDesc must have the same type name.
      assert(srcDesc.GetTypeNameForComparison(srcFieldDesc) == dstFieldDesc.GetTypeName());
      info.fInMemoryType = ColumnInMemoryType(dstFieldDesc.GetTypeName(), columnType);
      columns.emplace_back(info);
   }
 
   const auto &srcChildrenIds = srcFieldDesc.GetLinkIds();
   const auto &dstChildrenIds = dstFieldDesc.GetLinkIds();
   assert(srcChildrenIds.size() == dstChildrenIds.size());
   for (auto i = 0u; i < srcChildrenIds.size(); ++i) {
      const auto &srcChild = srcDesc.GetFieldDescriptor(srcChildrenIds[i]);
      const auto &dstChild = fDstDescriptor.GetFieldDescriptor(dstChildrenIds[i]);
      AddColumnsFromField(columns, srcDesc, colReprMappings, srcChild, dstChild, name);
   }
}
 
// Converts the fields comparison data to the corresponding column information.
// While doing so, it collects such information in `fColumnIdMap`, which is used by later calls to this
// function to map already-seen column names to their chosen outputId, type and so on.
RColumnInfoGroup RNTupleMergeStrategyDefault::GatherColumnInfos(const RDescriptorsComparison &descCmp,
                                                                const ROOT::RNTupleDescriptor &srcDesc)
{
   RColumnInfoGroup res;
   for (const ROOT::RFieldDescriptor *field : descCmp.fExtraDstFields) {
      AddColumnsFromField(res.fExtraDstColumns, fDstDescriptor, descCmp.fColReprMappings, *field, *field);
   }
   for (const auto &[srcField, dstField] : descCmp.fCommonFields) {
      AddColumnsFromField(res.fCommonColumns, srcDesc, descCmp.fColReprMappings, *srcField, *dstField);
   }
   return res;
}
 
// Generates default (zero) values for the given columns
ROOT::RResult<void> RNTupleMergeStrategyDefault::GenerateZeroPagesForColumns(size_t nEntriesToGenerate,
                                                                             std::span<const RColumnMergeInfo> columns,
                                                                             RSealedPageMergeData &sealedPageData)
{
   if (!nEntriesToGenerate)
      return ROOT::RResult<void>::Success();
 
   for (const auto &column : columns) {
      const ROOT::RFieldDescriptor *field = column.fParentFieldDescriptor;
 
      // Skip all auxiliary columns
      assert(!field->GetLogicalColumnIds().empty());
      if (field->GetLogicalColumnIds()[0] != column.fInputId)
         continue;
 
      // Check if this column is a child of a Collection or a Variant. If so, it has no data
      // and can be skipped.
      bool skipColumn = false;
      auto nRepetitions = std::max<std::uint64_t>(field->GetNRepetitions(), 1);
      for (auto parentId = field->GetParentId(); parentId != ROOT::kInvalidDescriptorId;) {
         const ROOT::RFieldDescriptor &parent = column.fParentNTupleDescriptor->GetFieldDescriptor(parentId);
         if (parent.GetStructure() == ROOT::ENTupleStructure::kCollection ||
             parent.GetStructure() == ROOT::ENTupleStructure::kVariant) {
            skipColumn = true;
            break;
         }
         nRepetitions *= std::max<std::uint64_t>(parent.GetNRepetitions(), 1);
         parentId = parent.GetParentId();
      }
      if (skipColumn)
         continue;
 
      const auto structure = field->GetStructure();
 
      if (structure == ROOT::ENTupleStructure::kStreamer) {
         return R__FAIL("Destination RNTuple contains a streamer field (" + field->GetFieldName() +
                        ") that is not present in one of the sources. "
                        "Creating a default value for a streamer field is ill-defined, therefore the merging "
                        "process will abort.");
      }
 
      // NOTE: we cannot have a Record here because it has no associated columns.
      R__ASSERT(structure == ROOT::ENTupleStructure::kCollection || structure == ROOT::ENTupleStructure::kVariant ||
                structure == ROOT::ENTupleStructure::kPlain);
 
      const auto &columnDesc = fDstDescriptor.GetColumnDescriptor(column.fOutputId);
      const auto colElement = RColumnElementBase::Generate(columnDesc);
      const auto nElements = nEntriesToGenerate * nRepetitions;
      const auto nBytesOnStorage = colElement->GetPackedSize(nElements);
      // TODO(gparolini): make this configurable
      constexpr auto kPageSizeLimit = 256 * 1024;
      // TODO(gparolini): consider coalescing the last page if its size is less than some threshold
      const size_t nPages = nBytesOnStorage / kPageSizeLimit + !!(nBytesOnStorage % kPageSizeLimit);
      for (size_t i = 0; i < nPages; ++i) {
         const auto pageSize = (i < nPages - 1) ? kPageSizeLimit : nBytesOnStorage - kPageSizeLimit * (nPages - 1);
         const auto checksumSize = RPageStorage::kNBytesPageChecksum;
         const auto bufSize = pageSize + checksumSize;
         assert(pageSize % colElement->GetSize() == 0);
         const auto nElementsPerPage = pageSize / colElement->GetSize();
         auto page = fPageAlloc->NewPage(colElement->GetSize(), nElementsPerPage);
         page.GrowUnchecked(nElementsPerPage);
         memset(page.GetBuffer(), 0, page.GetNBytes());
 
         auto &buffer = sealedPageData.fBuffers.emplace_back(new std::byte[bufSize]);
         RPageSink::RSealPageConfig sealConf;
         sealConf.fElement = colElement.get();
         sealConf.fPage = &page;
         sealConf.fBuffer = buffer.get();
         sealConf.fCompressionSettings = fMergeOpts.fCompressionSettings.value();
         sealConf.fWriteChecksum = fDestination.GetWriteOptions().GetEnablePageChecksums();
         auto sealedPage = RPageSink::SealPage(sealConf);
 
         sealedPageData.fPagesV.push_back({sealedPage});
         sealedPageData.fGroups.emplace_back(column.fOutputId, sealedPageData.fPagesV.back().cbegin(),
                                             sealedPageData.fPagesV.back().cend());
      }
   }
   return ROOT::RResult<void>::Success();
}
 
// Merges all columns appearing both in the source and destination RNTuples, just copying them if their
// compression matches ("fast merge") or by unsealing and resealing them with the proper compression.
ROOT::RResult<void> RNTupleMergeStrategyDefault::MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool,
                                                                    const ROOT::RClusterDescriptor &clusterDesc,
                                                                    std::span<RColumnMergeInfo> commonColumns,
                                                                    const RCluster::ColumnSet_t &commonColumnSet,
                                                                    RSealedPageMergeData &sealedPageData)
{
   const auto nCommonColumnsInCluster = commonColumnSet.size();
   assert(nCommonColumnsInCluster <= commonColumns.size());
 
   if (nCommonColumnsInCluster == 0)
      return ROOT::RResult<void>::Success();
 
   const RCluster *cluster = clusterPool.GetCluster(clusterDesc.GetId(), commonColumnSet);
   // we expect the cluster pool to contain the requested set of columns, since they were
   // validated by CompareDescriptorStructure() and MergeSourceClusters().
   assert(cluster);
 
   const std::uint32_t outCompression = fMergeOpts.fCompressionSettings.value();
 
   for (size_t colIdx = 0; colIdx < nCommonColumnsInCluster; ++colIdx) {
      const auto &column = commonColumns[colIdx];
      const auto &columnId = column.fInputId;
      R__ASSERT(clusterDesc.ContainsColumn(columnId));
 
      const auto &columnDesc = fSrcDescriptor->GetColumnDescriptor(columnId);
      const auto srcColElement = column.fInMemoryType
                                    ? ROOT::Internal::GenerateColumnElement(*column.fInMemoryType, columnDesc)
                                    : RColumnElementBase::Generate(columnDesc);
 
      // Now get the pages for this column in this cluster
      const auto &pages = clusterDesc.GetPageRange(columnId);
 
      RPageStorage::SealedPageSequence_t sealedPages;
      sealedPages.resize(pages.GetPageInfos().size());
 
      // Each column range potentially has a distinct compression settings
      const auto &columnRange = clusterDesc.GetColumnRange(columnId);
      assert(!columnRange.IsSuppressed());
      const auto colRangeCompressionSettings = columnRange.GetCompressionSettings().value();
 
      // Select "merging level". There are 2 levels, from fastest to slowest, depending on the case:
      // L1: compression and encoding of src and dest both match: we can simply copy the page
      // L2: compression of dest doesn't match the src we must recompress the page.
      // Note that in no case do we need to re-encode the page, as if the encoding differs we simply
      // append a new column representation to the field.
      const bool needsRecompressing = colRangeCompressionSettings != outCompression;
 
      if (needsRecompressing && fMergeOpts.fExtraVerbose) {
         R__LOG_INFO(NTupleMergeLog()) << "Recompressing column " << column.fColumnName
                                       << ": { compression: " << colRangeCompressionSettings << " => "
                                       << fMergeOpts.fCompressionSettings.value() << ", onDiskType: "
                                       << RColumnElementBase::GetColumnTypeName(
                                             srcColElement->GetIdentifier().fOnDiskType)
                                       << "}";
      }
 
      const size_t pageBufferBaseIdx = sealedPageData.fBuffers.size();
      // If the column range already has the right compression we don't need to allocate any new buffer, so we don't
      // bother reserving memory for them.
      if (needsRecompressing)
         sealedPageData.fBuffers.resize(sealedPageData.fBuffers.size() + pages.GetPageInfos().size());
 
      // If this column is deferred, we may need to fill "holes" until its real start. We fill any missing entry
      // with zeroes, like we do for extraDstColumns.
      // As an optimization, we don't do this for the first source (since we can rely on the FirstElementIndex and
      // deferred column mechanism in that case).
      // TODO: also avoid doing this if we added no real page of this column to the destination yet.
      if (columnDesc.GetFirstElementIndex() > clusterDesc.GetFirstEntryIndex() && fDestination.GetNEntries() > 0) {
         const auto nMissingEntries = columnDesc.GetFirstElementIndex() - clusterDesc.GetFirstEntryIndex();
         auto res = GenerateZeroPagesForColumns(nMissingEntries, {&column, 1}, sealedPageData);
         if (!res)
            return R__FORWARD_ERROR(res);
      }
 
      // Loop over the pages
      std::uint64_t pageIdx = 0;
      for (const auto &pageInfo : pages.GetPageInfos()) {
         assert(pageIdx < sealedPages.size());
         assert(sealedPageData.fBuffers.size() == 0 || pageIdx < sealedPageData.fBuffers.size());
         assert(pageInfo.GetLocator().GetType() != ROOT::RNTupleLocator::kTypePageZero);
 
         ROOT::Internal::ROnDiskPage::Key key{columnId, pageIdx};
         auto onDiskPage = cluster->GetOnDiskPage(key);
 
         if (!pageInfo.HasChecksum() && fDestination.GetWriteOptions().GetEnableSamePageMerging()) {
            R__LOG_WARNING(NTupleMergeLog())
               << "one or more pages were found without a checksum: disabling same-page merging.";
            const_cast<ROOT::RNTupleWriteOptions &>(fDestination.GetWriteOptions()).SetEnableSamePageMerging(false);
         }
 
         const auto checksumSize = pageInfo.HasChecksum() * RPageStorage::kNBytesPageChecksum;
         RPageStorage::RSealedPage &sealedPage = sealedPages[pageIdx];
         sealedPage.SetNElements(pageInfo.GetNElements());
         sealedPage.SetHasChecksum(pageInfo.HasChecksum());
         sealedPage.SetBufferSize(pageInfo.GetLocator().GetNBytesOnStorage() + checksumSize);
         sealedPage.SetBuffer(onDiskPage->GetAddress());
         // TODO(gparolini): more graceful error handling (skip the page?)
         sealedPage.VerifyChecksumIfEnabled().ThrowOnError();
         R__ASSERT(onDiskPage && (onDiskPage->GetSize() == sealedPage.GetBufferSize()));
 
         if (needsRecompressing) {
            const auto uncompressedSize = srcColElement->GetSize() * sealedPage.GetNElements();
            auto &buffer = sealedPageData.fBuffers[pageBufferBaseIdx + pageIdx];
            const auto bufSize = uncompressedSize + checksumSize;
            // NOTE: we currently allocate the max possible size for this buffer and don't shrink it afterward.
            // We might want to introduce an option that trades speed for memory usage and shrink the buffer to fit
            // the actual data size after recompressing.
            buffer = MakeUninitArray<std::byte>(bufSize);
 
            // clang-format off
            RTaskVisitor{fTaskGroup}(RChangeCompressionFunc{
               *srcColElement,
               outCompression,
               sealedPage,
               *fPageAlloc,
               buffer.get(),
               bufSize,
               fDestination.GetWriteOptions()
            });
            // clang-format on
         }
 
         ++pageIdx;
 
      } // end of loop over pages
 
      if (fTaskGroup)
         fTaskGroup->Wait();
 
      sealedPageData.fPagesV.push_back(std::move(sealedPages));
      sealedPageData.fGroups.emplace_back(column.fOutputId, sealedPageData.fPagesV.back().cbegin(),
                                          sealedPageData.fPagesV.back().cend());
   } // end loop over common columns
 
   return ROOT::RResult<void>::Success();
}
 
// Iterates over all clusters of `source` and merges their pages into `destination`.
// It is assumed that all columns in `commonColumns` are present (and compatible) in both the source and
// the destination's schemas.
// The pages may be "fast-merged" (i.e. simply copied with no decompression/recompression) if the target
// compression is unspecified or matches the original compression settings.
ROOT::RResult<void> RNTupleMergeStrategyDefault::MergeSourceClusters(RPageSource &source,
                                                                     std::span<RColumnMergeInfo> commonColumns,
                                                                     std::span<const RColumnMergeInfo> extraDstColumns)
{
   ROOT::Internal::RClusterPool clusterPool{source};
 
   std::vector<RColumnMergeInfo> missingColumns{extraDstColumns.begin(), extraDstColumns.end()};
 
   R__ASSERT(fSrcDescriptor->GetNClusters() == fSrcDescriptor->GetNActiveClusters());
   for (const auto &clusterDesc : fSrcDescriptor->GetActiveClusterIterable()) {
      const auto nClusterEntries = clusterDesc.GetNEntries();
      R__ASSERT(nClusterEntries > 0);
 
      // Deduce which columns are suppressed (cluster by cluster) by exclusion, as:
      // (columns in the columnIdMap) - (columns in commonColumns which are not suppressed).
      // Note that some suppressed columns may not be in commonColumns because they might not appear at all in the
      // current source.
      FieldCollectionMap_t<ROOT::DescriptorId_t> columnsInCluster;
      using ColumnHandle_t = ROOT::Internal::RPageStorage::ColumnHandle_t;
 
      // NOTE: `commonColumns` contains all columns that appear *somewhere* both in the src and in the dst.
      // Just because a column is in `commonColumns` it doesn't mean that each cluster in the source contains
      // it, as it may be a deferred column that only has real data in a future cluster. We need to figure out which
      // columns are actually present in this cluster so we only merge their pages (the missing columns are handled
      // by synthesizing zero pages - see below).
 
      // Convert columns to a ColumnSet for the ClusterPool query
      RCluster::ColumnSet_t commonColumnSet;
      // Collect all common columns appearing in this cluster into commonColumnSet and reorganize commonColumns so
      // that those columns are at the start of it (whereas missing columns are at its end).
      // NOTE: it's fine if this scrambles the order of columns: the RNTupleSerializer will sort them by physical ID.
      int nCommonColumnsInCluster = 0;
      std::partition(commonColumns.begin(), commonColumns.end(), [&](const auto &column) {
         if (clusterDesc.ContainsColumn(column.fInputId)) {
            const auto &colRange = clusterDesc.GetColumnRange(column.fInputId);
            ++nCommonColumnsInCluster;
            columnsInCluster[column.fParentFieldDescriptor].push_back(column.fOutputId);
            if (!colRange.IsSuppressed()) {
               commonColumnSet.emplace(column.fInputId);
               return true;
            }
            fDestination.CommitSuppressedColumn(ColumnHandle_t{column.fOutputId});
         }
         return false;
      });
 
      // Commit all suppressed columns.
      // This is a fairly involved operation, as we need to commit all known columns that:
      // a) do not appear in extraDstColumns (those are "missing", not suppressed), and
      // b) do not appear in commonColumnSet (those are the active columns).
      // Not that these may or may not appear in commonColumns as suppressed columns, since they may or may not be
      // present in the current source.
      // The only way to find all the columns is to go and get them from fColumnIdMap, which keeps track of every
      // column we added to the destination so far. However, since it also contains the extraDstColumns, we need to
      // specifically only query those columns that belong to a field that has at least 1 column in commonColumns
      // (remember that commonColumns contains all columns associated to the common fields for this source).
      for (const auto &[fieldDesc, columnIds] : columnsInCluster) {
         const auto &fieldFQName = fSrcDescriptor->GetQualifiedFieldName(fieldDesc->GetId());
         const auto cardinality = fieldDesc->GetColumnCardinality();
         for (auto i = 0u; i < fieldDesc->GetLogicalColumnIds().size(); ++i) {
            const auto colIndex = i % cardinality;
            const auto reprIndex = i / cardinality;
            const auto colName = "." + fieldFQName + '.' + std::to_string(colIndex) + '.' + std::to_string(reprIndex);
            const auto colIt = fColumnIdMap.find(colName);
            assert(colIt != fColumnIdMap.end());
            const auto colOutId = colIt->second.fColumnId;
            if (std::find(columnIds.begin(), columnIds.end(), colOutId) == columnIds.end()) {
               fDestination.CommitSuppressedColumn(ColumnHandle_t{colOutId});
            }
         }
      }
 
      RSealedPageMergeData sealedPageData;
      auto res = MergeCommonColumns(clusterPool, clusterDesc, commonColumns, commonColumnSet, sealedPageData);
      if (!res)
         return R__FORWARD_ERROR(res);
 
      // Generate zero pages for the missing columns.
      // For each cluster, the "missing columns" are the union of the extraDstColumns and the common columns
      // that are not present in the cluster.
      // Note that this does NOT include suppressed columns, for which no pages are synthesized.
      missingColumns.resize(extraDstColumns.size()); // NOTE: this clears all common columns of the previous cluster
      for (size_t i = nCommonColumnsInCluster; i < commonColumns.size(); ++i)
         missingColumns.push_back(commonColumns[i]);
 
      res = GenerateZeroPagesForColumns(nClusterEntries, missingColumns, sealedPageData);
      if (!res)
         return R__FORWARD_ERROR(res);
 
      // Commit the pages and the clusters
      fDestination.CommitSealedPageV(sealedPageData.fGroups);
      fDestination.CommitCluster(nClusterEntries);
   }
 
   // TODO(gparolini): when we get serious about huge file support (>~ 100GB) we might want to check here
   // the size of the running page list and commit a cluster group when it exceeds some threshold,
   // which would prevent the page list from getting too large.
   // However, as of today, we aren't really handling such huge files, and even relatively big ones
   // such as the CMS dataset have a page list size of about only 2 MB.
   // So currently we simply merge all cluster groups into one.
   return ROOT::RResult<void>::Success();
}
 
ROOT::RResult<void> RNTupleMergeStrategyDefault::MergeSource(RPageSource &source, RDescriptorsComparison &descCmp)
{
   //// Extend columns if needed
   if (!descCmp.fColReprExtensions.empty()) {
      for (const auto &field : descCmp.fExtraDstFields) {
         if (field->IsProjectedField())
            fProjectionPointees[field->GetProjectionSourceId()].push_back(field);
      }
 
      // We need to extend the columns in the proper order, i.e. so that they appear in the same order as
      // their first representation. This is to ensure that the pages we write to the cluster are in a consistent
      // order as their column descriptors. The page creation order is determined by the order of
      // columnInfos.fCommonColumns, which in turn depends on the common fields order (see GatherColumnInfos).
      // XXX: do we need this separate sort step? Why not just create this vector directly in
      // CompareDescriptorStructure?
      std::vector<std::pair<const ROOT::RFieldDescriptor *, std::vector<RColReprExtension>>> colExtensions;
      colExtensions.reserve(descCmp.fColReprExtensions.size());
      for (const auto &commonField : descCmp.fCommonFields) {
         const auto *field = commonField.fDst;
         AddColumnExtensionsInFieldOrder(*field, fDstDescriptor, descCmp.fColReprExtensions, colExtensions,
                                         fProjectionPointees);
      }
      for (const auto &field : descCmp.fExtraSrcFields) {
         if (field->IsProjectedField())
            fProjectionPointees[field->GetProjectionSourceId()].push_back(field);
      }
 
      for (const auto &[fieldDesc, extensions] : colExtensions) {
         auto &mappings = descCmp.fColReprMappings[fieldDesc];
         for (const auto &extension : extensions) {
            const auto firstColumnId = fDestination.AddColumnRepresentation(*fieldDesc, extension.fSourceRepr,
                                                                            extension.fOrigFirstElementIndex);
 
            // When adding new column representations to an existing field which is the source of some projected
            // fields, we need to also add new alias columns to those fields so that they can point to the proper
            // representation.
            if (auto it = fProjectionPointees.find(fieldDesc->GetId()); it != fProjectionPointees.end()) {
               for (const auto &projection : it->second) {
                  for (auto colIdx = 0u; colIdx < extension.fSourceRepr.size(); ++colIdx)
                     fDestination.AddAliasColumn(fDstDescriptor, *projection, firstColumnId + colIdx);
               }
            }
            mappings.push_back(extension);
         }
      }
   }
 
   // handle extra dst fields & common fields
   auto columnInfos = GatherColumnInfos(descCmp, *fSrcDescriptor);
   auto res = MergeSourceClusters(source, columnInfos.fCommonColumns, columnInfos.fExtraDstColumns);
   return res;
}
 
// =========================================================================================
//
// MergeStrategySlow
//
// =========================================================================================
namespace ROOT::Internal {
// Strategy used for L4 merging (https://github.com/root-project/root/blob/master/tree/ntuple/doc/Merging.md#glossary).
// This class must be in ROOT::Experimental::Internal because it befriends ROOT::REntry
class RNTupleMergeStrategySlow final : public RNTupleMergeStrategy {
   std::unique_ptr<ROOT::REntry> fDstEntry;
   std::unique_ptr<RNTupleMergeActions> fActions = std::make_unique<RNTupleMergeActions>();
   RNTupleMergeActionContext fMergeContext;
 
public:
   RNTupleMergeStrategySlow(std::unique_ptr<RNTupleModel> dstModel, ROOT::Internal::RPagePersistentSink &dst,
                            const RNTupleMergeOptions &mergeOpts,
                            std::unique_ptr<RNTupleMergeActions> actions = std::make_unique<RNTupleMergeActions>())
      : RNTupleMergeStrategy(std::move(dstModel), dst, mergeOpts), fActions(std::move(actions))
   {
      // dstModel is non null iff we are doing an incremental merge. Currently we don't support it.
      assert(!dstModel);
   }
 
   ROOT::NTupleSize_t GetNDstEntries() const final
   {
      // During slow merging clusters are not necessarily committed at source boundary, so we need to
      // manually track this number.
      return fMergeContext.fDestinationEntryIndex;
   }
 
   void InitSource(RPageSource &source) final
   {
      source.Attach();
      fSrcDescriptor = &source.GetSharedDescriptorGuard().GetConstRef();
 
      if (!fDstModel) {
         ROOT::RNTupleDescriptor::RCreateModelOptions cmOpts;
         cmOpts.SetCreateBare(true);
         cmOpts.SetReconstructProjections(true);
         fDstModel = source.GetSharedDescriptorGuard()->CreateModel(cmOpts);
         fDestination.Init(*fDstModel);
         fDstEntry = fDstModel->CreateEntry();
      }
   }
 
   [[nodiscard]] ROOT::RResult<void> ExtendDestinationModel(RDescriptorsComparison &descCmp) final
   {
      // late model extension for all fExtraSrcFields in Union mode
      auto res = ::ExtendDestinationModel(descCmp, *fDstModel, fDestination, *fSrcDescriptor,
                                          fMergeContext.fDestinationEntryIndex, fMergeOpts.fExtraVerbose,
                                          /*forSlowMerge=*/true);
      // recreate the entry since the model has changed
      fDstEntry = fDstModel->CreateEntry();
      return res;
   }
 
   [[nodiscard]] ROOT::RResult<void> MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) final;
 
   void BeforeCommit() final
   {
      while (true) {
         auto res = fActions->BeforeCommit(*fDstEntry, fMergeContext);
         if (res == ENTupleMergeActionResult::kDrop)
            break;
         fMergeContext.fDestinationEntryIndex += 1;
         fDstEntry->Append();
         if (res == ENTupleMergeActionResult::kEmit)
            break;
      }
 
      // TODO: this should happen more often than once at the end and it should depend on the target cluster size...
      if (fMergeContext.fDestinationEntryIndex > 0) {
         fDestination.CommitCluster(fMergeContext.fDestinationEntryIndex);
      }
   }
};
} // namespace ROOT::Internal
 
ROOT::RResult<void> RNTupleMergeStrategySlow::MergeSource(RPageSource &source, RDescriptorsComparison &descCmp)
{
   const auto &dstDesc = fDestination.GetDescriptor();
 
   // Make sure that types of common fields match
   for (const auto &[srcField, dstField] : descCmp.fCommonFields) {
      const auto srcType = fSrcDescriptor->GetTypeNameForComparison(*srcField);
      const auto dstType = dstDesc.GetTypeNameForComparison(*dstField);
      if (srcType != dstType) {
         return R__FAIL("Field `" + fSrcDescriptor->GetQualifiedFieldName(srcField->GetId()) +
                        "` has a type incompatible with a previously-seen field with the same name: (old: " + srcType +
                        ", new: " + dstType + ")");
      }
   }
 
   // Impose the destination model on the source. This will ensure we properly schema-evolve the fields if
   // necessary (e.g. if this source has fewer/reordered fields than the destination)
   auto srcModel = fDstModel->Clone();
 
   // NOTE: this code is almost the same as RNTupleReader::ConnectModel().
   // Perhaps it can be made common?
   {
      auto &fieldZero = ROOT::Internal::GetFieldZeroOfModel(*srcModel);
      ROOT::Internal::SetAllowFieldSubstitutions(fieldZero, true);
      // We must not use the descriptor guard to prevent recursive locking in field.ConnectPageSource
      ROOT::DescriptorId_t fieldZeroId = fSrcDescriptor->GetFieldZeroId();
      fieldZero.SetOnDiskId(fieldZeroId);
      const auto resetFieldIdAndColumnRepresentatives = [](ROOT::RFieldBase &field, const auto &recursiveCall) -> void {
         field.SetOnDiskId(ROOT::kInvalidDescriptorId);
         field.SetColumnRepresentatives({});
         for (auto *f : field.GetMutableSubfields()) {
            recursiveCall(*f, recursiveCall);
         }
      };
      // Iterate only over fieldZero's direct subfields; their descendants are recursively handled in
      // RFieldBase::ConnectPageSource
      for (auto *field : fieldZero.GetMutableSubfields()) {
         // We need to reset column reprs otherwise ConnectPageSource fails in case we are changing the column reprs
         // due to AutoAdjustColumnRepresentatives.
         // We also need to reset on disk ids otherwise fields get matched with the wrong ones using the writer's ids
         // (which differ from the reader's)
         resetFieldIdAndColumnRepresentatives(*field, resetFieldIdAndColumnRepresentatives);
 
         if (field->GetOnDiskId() == ROOT::kInvalidDescriptorId)
            field->SetOnDiskId(fSrcDescriptor->FindFieldId(field->GetFieldName(), fieldZeroId));
 
         // If the OnDiskId is still invalid at this point it means this is an extraDstField
         // and we shouldn't/cannot connect it.
         if (field->GetOnDiskId() != ROOT::kInvalidDescriptorId)
            ROOT::Internal::CallConnectPageSourceOnField(*field, source);
         else
            ROOT::Internal::CallSetArtificialOnField(*field);
      }
      ROOT::Internal::SetAllowFieldSubstitutions(fieldZero, false);
   }
 
   auto srcEntry = srcModel->CreateBareEntry();
 
   // Create and connect fields to source and sink and bind them together.
   // This causes a read on the src entry to populate the same memory used by the dst entry to write.
   for (const auto &[srcFieldDesc, _dstFieldDesc] : descCmp.fCommonFields) {
      if (!srcFieldDesc->IsProjectedField()) {
         // Reset even common fields to their default value, since they may be Record fields with missing subfields
         // in the source (in which case we won't overwrite them with new values and they'd keep the old ones)
         fDstEntry->EmplaceNewValue(srcFieldDesc->GetFieldName());
         srcEntry->BindValue(srcFieldDesc->GetFieldName(), fDstEntry->GetPtr<void>(srcFieldDesc->GetFieldName()));
      }
   }
   // reset the extraDstfields, to make sure any subfield that's missing from the current source
   // is default-initialized (rather than keeping its previous value).
   for (const auto *field : descCmp.fExtraDstFields) {
      if (!field->IsProjectedField()) {
         fDstEntry->EmplaceNewValue(field->GetFieldName());
      }
   }
 
   // Do the actual data reading/writing
   for (auto idx = 0u; idx < fSrcDescriptor->GetNEntries(); ++idx) {
      srcEntry->Read(idx);
      fMergeContext.fSourceEntryIndex = idx;
      while (true) {
         auto res = fActions->OnFill(*fDstEntry, fMergeContext);
         if (res == ENTupleMergeActionResult::kDrop)
            break;
         fMergeContext.fDestinationEntryIndex += 1;
         fDstEntry->Append();
         if (res == ENTupleMergeActionResult::kEmit)
            break;
      }
   }
 
   while (true) {
      auto res = fActions->OnSourceEnd(*fDstEntry, fMergeContext);
      if (res == ENTupleMergeActionResult::kDrop)
         break;
      fMergeContext.fDestinationEntryIndex += 1;
      fDstEntry->Append();
      if (res == ENTupleMergeActionResult::kEmit)
         break;
   }
 
   if (fMergeContext.fDestinationEntryIndex > 0) {
      for (auto &field : ROOT::Internal::GetFieldZeroOfModel(*fDstModel)) {
         ROOT::Internal::CallFlushColumnsOnField(field);
      }
   }
 
   return ROOT::RResult<void>::Success();
}
 
// =========================================================================================
//
// RNTupleMerger
//
// =========================================================================================
RNTupleMerger::RNTupleMerger(std::unique_ptr<ROOT::Internal::RPagePersistentSink> destination,
                             std::unique_ptr<ROOT::RNTupleModel> model)
   : fDestination(std::move(destination)), fModel(std::move(model))
{
   R__ASSERT(fDestination);
}
 
RNTupleMerger::RNTupleMerger(std::unique_ptr<ROOT::Internal::RPagePersistentSink> destination)
   : ROOT::Internal::RNTupleMerger(std::move(destination), nullptr)
{
}
 
ROOT::RResult<void> RNTupleMerger::Merge(std::span<RPageSource *> sources, const RNTupleMergeOptions &mergeOptsIn)
{
   RNTupleMergeOptions mergeOpts = mergeOptsIn;
 
   assert(fDestination);
 
   // TODO(gparolini): this is here to allow testing of the slow merging path, it will disappear in the future.
   constexpr bool doSlowMerge = false;
 
   // Set compression settings if unset and verify it's compatible with the sink
   {
      const auto dstCompSettings = fDestination->GetWriteOptions().GetCompression();
      if (!mergeOpts.fCompressionSettings) {
         mergeOpts.fCompressionSettings = dstCompSettings;
      } else if (*mergeOpts.fCompressionSettings != dstCompSettings) {
         return R__FAIL(std::string("The compression given to RNTupleMergeOptions is different from that of the "
                                    "sink! (opts: ") +
                        std::to_string(*mergeOpts.fCompressionSettings) + ", sink: " + std::to_string(dstCompSettings) +
                        ") This is currently unsupported.");
      }
   }
 
   // we should have a model if and only if the destination is initialized.
   if (!!fModel != fDestination->IsInitialized()) {
      return R__FAIL(
         "passing an already-initialized destination to RNTupleMerger::Merge (i.e. trying to do incremental "
         "merging) can only be done by providing a valid ROOT::RNTupleModel when constructing the RNTupleMerger.");
   }
 
   if (doSlowMerge && fModel) {
      return R__FAIL("incremental slow merging is currently unsupported.");
   }
 
   // NOTE: don't wrap this in a do {} while (0)! It uses continue!
#define SKIP_OR_ABORT(errMsg)                                                      \
   if (mergeOpts.fErrBehavior == ENTupleMergeErrBehavior::kSkip) {                 \
      R__LOG_WARNING(NTupleMergeLog()) << "Skipping RNTuple due to: " << (errMsg); \
      continue;                                                                    \
   } else {                                                                        \
      return R__FAIL(errMsg);                                                      \
   }
 
   auto strat = doSlowMerge ? std::unique_ptr<RNTupleMergeStrategy>(
                                 new RNTupleMergeStrategySlow(std::move(fModel), *fDestination, mergeOpts))
                            : std::unique_ptr<RNTupleMergeStrategy>(
                                 new RNTupleMergeStrategyDefault(std::move(fModel), *fDestination, mergeOpts));
   struct RAttrSetMergeData {
      std::unique_ptr<ROOT::Internal::RPagePersistentSink> fSink;
      RNTupleMergeStrategySlow fStrategy;
 
      RAttrSetMergeData(std::unique_ptr<ROOT::Internal::RPagePersistentSink> attrSink,
                        const RNTupleMergeStrategy &parentStrategy, const RNTupleMergeOptions &mergeOpts,
                        ROOT::NTupleSize_t initialStartOffset)
         : fSink(std::move(attrSink)),
           fStrategy(nullptr, *fSink, mergeOpts,
                     std::make_unique<RNTupleAttrMergeActions>(parentStrategy, initialStartOffset))
      {
      }
   };
   // Map { attrSetName => mergeData }
   std::unordered_map<std::string, RAttrSetMergeData> attrSetsData;
 
   // Merge main loop
   for (RPageSource *source : sources) {
      strat->InitSource(*source);
      {
         auto srcDescriptor = source->GetSharedDescriptorGuard();
         if (srcDescriptor->GetVersion() > ROOT::RNTuple::GetCurrentVersion()) {
            if (mergeOpts.fVersionBehavior == ENTupleMergeVersionBehavior::kWarnOnHigherVersion) {
               R__LOG_WARNING(NTupleMergeLog())
                  << "RNTuple '" << srcDescriptor->GetName()
                  << "' has a higher format version than the latest supported by this version "
                     "of ROOT. Merging will work but some features may be dropped.";
            } else {
               return R__FAIL("RNTuple '" + srcDescriptor->GetName() +
                              "' has a higher format version than the latest supported by this version. Refusing to "
                              "merge, since RNTupleMergeOptions::fVersionBehavior is set to AbortOnHigherVersion.");
            }
         }
 
         for (const auto &extraTypeInfoDesc : srcDescriptor->GetExtraTypeInfoIterable())
            fDestination->UpdateExtraTypeInfo(extraTypeInfoDesc);
      }
 
      auto descCmpRes =
         CompareDescriptorStructure(fDestination->GetDescriptor(), source->GetSharedDescriptorGuard().GetConstRef());
      if (!descCmpRes) {
         SKIP_OR_ABORT(std::string("Source RNTuple has an incompatible schema with the destination:\n") +
                       descCmpRes.GetError()->GetReport())
      }
      auto descCmp = descCmpRes.Unwrap();
      // TODO: switch to slow merge if we cannot "fast" merge
      if (!doSlowMerge && !descCmp.fFastMergeErrors.empty()) {
         SKIP_OR_ABORT(std::string("Source RNTuple has an incompatible schema with the destination:\n") +
                       descCmp.fFastMergeErrors)
      }
 
      // If the current source is missing some fields and we're not in Union mode, error
      // (if we are in Union mode, MergeSourceClusters will fill the missing fields with default values).
      if (mergeOpts.fMergingMode != ENTupleMergingMode::kUnion && !descCmp.fExtraDstFields.empty()) {
         std::string msg = "Source RNTuple is missing the following fields:";
         for (const auto *field : descCmp.fExtraDstFields) {
            msg += "\n  " + field->GetFieldName() + " : " + field->GetTypeName();
         }
         SKIP_OR_ABORT(msg)
      }
 
      // handle extra src fields
      if (!descCmp.fExtraSrcFields.empty()) {
         if (mergeOpts.fMergingMode == ENTupleMergingMode::kUnion) {
            // late model extension for all fExtraSrcFields in Union mode
            auto res = strat->ExtendDestinationModel(descCmp);
            if (!res)
               return R__FORWARD_ERROR(res);
         } else if (mergeOpts.fMergingMode == ENTupleMergingMode::kStrict) {
            // If the current source has extra fields and we're in Strict mode, error
            std::string msg = "Source RNTuple has extra fields that the destination RNTuple doesn't have:";
            for (const auto *field : descCmp.fExtraSrcFields) {
               msg += "\n  " + field->GetFieldName() + " : " + field->GetTypeName();
            }
            SKIP_OR_ABORT(msg)
         }
      }
 
      const auto prevNEntries = strat->GetNDstEntries();
 
      auto res = strat->MergeSource(*source, descCmp);
      if (!res)
         return R__FORWARD_ERROR(res);
 
      // Slow-merge attributes
      for (const auto &attrSetDesc : source->GetSharedDescriptorGuard()->GetAttrSetIterable()) {
         // Initialize output sink and strategy if it's the first time we encounter this attribute set name
         auto attrSetDataIt = attrSetsData.find(attrSetDesc.GetName());
         if (attrSetDataIt == attrSetsData.end()) {
            auto sinkOpts = RNTupleWriteOptions();
            sinkOpts.SetCompression(mergeOpts.fCompressionSettings.value());
            auto sink =
               std::unique_ptr<ROOT::Internal::RPagePersistentSink>(static_cast<ROOT::Internal::RPagePersistentSink *>(
                  fDestination->CloneAsHidden(attrSetDesc.GetName(), sinkOpts).release()));
            auto [insertedIt, inserted] =
               attrSetsData.try_emplace(attrSetDesc.GetName(), std::move(sink), *strat, mergeOpts, prevNEntries);
            assert(inserted);
            attrSetDataIt = insertedIt;
         }
         auto &attrSetData = attrSetDataIt->second;
 
         const auto attrLink =
            ROOT::Internal::RNTupleLink{attrSetDesc.GetAnchorLocator(), attrSetDesc.GetAnchorLength()};
         auto attrSource = source->OpenWithDifferentAnchor(attrLink);
         auto &attrStrat = attrSetData.fStrategy;
         attrStrat.InitSource(*attrSource);
 
         descCmpRes = CompareDescriptorStructure(attrSetData.fSink->GetDescriptor(),
                                                 attrSource->GetSharedDescriptorGuard().GetConstRef(),
                                                 /*forAttributeMerging=*/true);
         if (!descCmpRes) {
            SKIP_OR_ABORT(std::string("Source RNTuple has an incompatible schema with the destination:\n") +
                          descCmpRes.GetError()->GetReport())
         }
         descCmp = descCmpRes.Unwrap();
 
         if (!descCmp.fExtraSrcFields.empty() || !descCmp.fExtraSrcSubfields.empty()) {
            auto edmRes = attrStrat.ExtendDestinationModel(descCmp);
            if (!edmRes)
               return R__FORWARD_ERROR(edmRes);
         }
 
         auto mergeSourceRes = attrStrat.MergeSource(*attrSource, descCmp);
         if (!mergeSourceRes)
            return R__FORWARD_ERROR(mergeSourceRes);
      }
   } // end loop over sources
 
   // Commit the output
   for (auto &[attrSetName, attrSetData] : attrSetsData) {
      attrSetData.fStrategy.BeforeCommit();
      attrSetData.fSink->CommitClusterGroup();
      auto anchorLink = attrSetData.fSink->CommitDataset();
      fDestination->CommitAttributeSet(attrSetName, anchorLink);
   }
 
   strat->BeforeCommit();
 
   // Commit the output
   if (fDestination->GetNEntries() == 0)
      R__LOG_WARNING(NTupleMergeLog()) << "Output RNTuple '" << fDestination->GetNTupleName() << "' has no entries.";
 
   fDestination->CommitClusterGroup();
   fDestination->CommitDataset();
 
   return RResult<void>::Success();
}