RNTupleMerger.cxx

Go to the documentation of this file.

/// \file RNTupleMerger.cxx
/// \ingroup NTuple ROOT7
/// \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/RNTupleUtil.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 <TROOT.h>
#include <TFileMergeInfo.h>
#include <TError.h>
#include <TFile.h>
#include <TKey.h>
 
#include <algorithm>
#include <deque>
#include <inttypes.h> // for PRIu64
#include <unordered_map>
#include <vector>
 
using namespace ROOT::Experimental;
using namespace ROOT::Experimental::Internal;
 
// Entry point for TFileMerger. Internally calls RNTupleMerger::Merge().
Long64_t 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) {
      Error("RNTuple::Merge", "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) {
      Error("RNTuple::Merge", "Second input parameter should be a TFile, but it's a %s.", secondArg->ClassName());
      return -1;
   }
 
   // Check if the output file already has a key with that name
   TKey *outKey = outFile->FindKey(ntupleName.c_str());
   RNTuple *outNTuple = nullptr;
   if (outKey) {
      outNTuple = outKey->ReadObject<RNTuple>();
      if (!outNTuple) {
         Error("RNTuple::Merge", "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.
   }
 
   // The "fast" option is present if and only if we don't want to change compression.
   const int compression =
      mergeInfo->fOptions.Contains("fast") ? kUnknownCompressionSettings : outFile->GetCompressionSettings();
 
   RNTupleWriteOptions writeOpts;
   writeOpts.SetUseBufferedWrite(false);
   if (compression != kUnknownCompressionSettings)
      writeOpts.SetCompression(compression);
   auto destination = std::make_unique<RPageSinkFile>(ntupleName, *outFile, writeOpts);
 
   // If we already have an existing RNTuple, copy over its descriptor to support incremental merging
   if (outNTuple) {
      auto source = RPageSourceFile::CreateFromAnchor(*outNTuple);
      source->Attach();
      auto desc = source->GetSharedDescriptorGuard();
      destination->InitFromDescriptor(desc.GetRef());
   }
 
   // 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);
      RNTuple *anchor = inFile ? inFile->Get<RNTuple>(ntupleName.c_str()) : nullptr;
      if (!anchor) {
         Error("RNTuple::Merge", "Failed to retrieve RNTuple anchor named '%s' from file '%s'", ntupleName.c_str(),
               inFile->GetName());
         return -1;
      }
      sources.push_back(RPageSourceFile::CreateFromAnchor(*anchor));
   }
 
   // Interface conversion
   sourcePtrs.reserve(sources.size());
   for (const auto &s : sources) {
      sourcePtrs.push_back(s.get());
   }
 
   // Now merge
   RNTupleMerger merger;
   RNTupleMergeOptions mergerOpts;
   mergerOpts.fCompressionSettings = compression;
   merger.Merge(sourcePtrs, *destination).ThrowOnError();
 
   // Provide the caller with a merged anchor object (even though we've already
   // written it).
   *this = *outFile->Get<RNTuple>(ntupleName.c_str());
 
   return 0;
} catch (const RException &ex) {
   Error("RNTuple::Merge", "Exception thrown while merging: %s", ex.what());
   return -1;
}
 
namespace {
// Functor used to change the compression of a page to `options.fCompressionSettings`.
struct RChangeCompressionFunc {
   DescriptorId_t fOutputColumnId;
 
   const RColumnElementBase &fSrcColElement;
   const RColumnElementBase &fDstColElement;
   const RNTupleMergeOptions &fMergeOptions;
 
   RPageStorage::RSealedPage &fSealedPage;
   RPageAllocator &fPageAlloc;
   std::uint8_t *fBuffer;
 
   void operator()() const
   {
      auto page = RPageSource::UnsealPage(fSealedPage, fSrcColElement, fOutputColumnId, fPageAlloc).Unwrap();
      RPageSink::RSealPageConfig sealConf;
      sealConf.fElement = &fDstColElement;
      sealConf.fPage = &page;
      sealConf.fBuffer = fBuffer;
      sealConf.fCompressionSetting = fMergeOptions.fCompressionSettings;
      sealConf.fWriteChecksum = fSealedPage.GetHasChecksum();
      auto refSealedPage = RPageSink::SealPage(sealConf);
      fSealedPage = refSealedPage;
   }
};
 
struct RDescriptorsComparison {
   std::vector<const RFieldDescriptor *> fExtraDstFields;
   std::vector<const RFieldDescriptor *> fExtraSrcFields;
   std::vector<const RFieldDescriptor *> fCommonFields;
};
 
struct RColumnOutInfo {
   DescriptorId_t fColumnId;
   EColumnType fColumnType;
};
 
// { fully.qualified.fieldName.colInputId => colOutputInfo }
using ColumnIdMap_t = std::unordered_map<std::string, RColumnOutInfo>;
 
struct RColumnInfoGroup {
   std::vector<RColumnInfo> fExtraDstColumns;
   std::vector<RColumnInfo> fCommonColumns;
};
 
} // namespace
 
// These structs cannot be in the anon namespace becase they're used in RNTupleMerger's private interface.
namespace ROOT::Experimental::Internal {
struct RColumnInfo {
   // 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;
   DescriptorId_t fInputId;
   DescriptorId_t fOutputId;
   EColumnType fColumnType;
   const RFieldDescriptor *fParentField;
};
 
// Data related to a single call of RNTupleMerger::Merge()
struct RNTupleMergeData {
   std::span<RPageSource *> fSources;
   RPageSink &fDestination;
   const RNTupleMergeOptions &fMergeOpts;
   const RNTupleDescriptor &fDstDescriptor;
   const RNTupleDescriptor *fSrcDescriptor = nullptr;
 
   std::vector<RColumnInfo> fColumns;
   ColumnIdMap_t fColumnIdMap;
 
   NTupleSize_t fNumDstEntries = 0;
 
   RNTupleMergeData(std::span<RPageSource *> sources, RPageSink &destination, const RNTupleMergeOptions &mergeOpts)
      : fSources{sources}, fDestination{destination}, fMergeOpts{mergeOpts}, fDstDescriptor{destination.GetDescriptor()}
   {
   }
};
 
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::uint8_t[]>> fBuffers;
};
} // namespace ROOT::Experimental::Internal
 
/// 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 RResult<RDescriptorsComparison>
CompareDescriptorStructure(const RNTupleDescriptor &dst, const RNTupleDescriptor &src)
{
   // 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)
 
   std::vector<std::string> errors;
   RDescriptorsComparison res;
 
   struct RCommonField {
      const RFieldDescriptor *fDst;
      const RFieldDescriptor *fSrc;
   };
   std::vector<RCommonField> commonFields;
 
   for (const auto &dstField : dst.GetTopLevelFields()) {
      const auto srcFieldId = src.FindFieldId(dstField.GetFieldName());
      if (srcFieldId != kInvalidDescriptorId) {
         const auto &srcField = src.GetFieldDescriptor(srcFieldId);
         commonFields.push_back({&dstField, &srcField});
      } else {
         res.fExtraDstFields.emplace_back(&dstField);
      }
   }
   for (const auto &srcField : src.GetTopLevelFields()) {
      const auto dstFieldId = dst.FindFieldId(srcField.GetFieldName());
      if (dstFieldId == kInvalidDescriptorId)
         res.fExtraSrcFields.push_back(&srcField);
   }
 
   // Check compatibility of common fields
   for (const auto &field : commonFields) {
      // 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
      bool projCompatible = field.fSrc->IsProjectedField() == field.fDst->IsProjectedField();
      if (!projCompatible) {
         std::stringstream ss;
         ss << "Field `" << fieldName << "` is incompatible with previously-seen field with that name because the "
            << (field.fSrc->IsProjectedField() ? "new" : "old") << " one is projected and the other isn't";
         errors.push_back(ss.str());
      } else if (field.fSrc->IsProjectedField()) {
         // if both fields are projected, verify that they point to the same real field
         const auto srcName = src.GetQualifiedFieldName(field.fSrc->GetProjectionSourceId());
         const auto dstName = dst.GetQualifiedFieldName(field.fDst->GetProjectionSourceId());
         if (srcName != dstName) {
            std::stringstream ss;
            ss << "Field `" << fieldName
               << "` is projected to a different field than a previously-seen field with the same name (old: " << dstName
               << ", new: " << srcName << ")";
            errors.push_back(ss.str());
         }
      }
 
      // Require that fields types match
      // TODO(gparolini): allow non-identical but compatible types
      const auto &srcTyName = field.fSrc->GetTypeName();
      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());
      }
 
      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());
      }
 
      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());
      }
   }
 
   std::string errMsg;
   for (const auto &err : errors)
      errMsg += std::string("\n  * ") + err;
 
   if (!errMsg.empty())
      errMsg = errMsg.substr(1); // strip initial newline
 
   if (errMsg.length())
      return R__FAIL(errMsg);
 
   res.fCommonFields.reserve(commonFields.size());
   for (const auto &[_, srcField] : commonFields) {
      res.fCommonFields.emplace_back(srcField);
   }
 
   // TODO(gparolini): we should exhaustively check the field tree rather than just the top level fields,
   // in case the user forgets to change the version number on one field.
 
   return RResult(res);
}
 
// Applies late model extension to `destination`, adding all `newFields` to it.
static void ExtendDestinationModel(std::span<const RFieldDescriptor *> newFields, RNTupleModel &dstModel,
                                   RNTupleMergeData &mergeData)
{
   assert(newFields.size() > 0); // no point in calling this with 0 new cols
 
   dstModel.Unfreeze();
   RNTupleModelChangeset changeset{dstModel};
 
   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() + '`';
   });
   Info("RNTuple::Merge", "%s: adding %s to the destination model (entry #%" PRIu64 ").", msg.c_str(),
        (newFields.size() > 1 ? "them" : "it"), mergeData.fNumDstEntries);
 
   changeset.fAddedFields.reserve(newFields.size());
   for (const auto *fieldDesc : newFields) {
      auto field = fieldDesc->CreateField(*mergeData.fSrcDescriptor);
      if (fieldDesc->IsProjectedField())
         changeset.fAddedProjectedFields.emplace_back(field.get());
      else
         changeset.fAddedFields.emplace_back(field.get());
      changeset.fModel.AddField(std::move(field));
   }
   dstModel.Freeze();
   mergeData.fDestination.UpdateSchema(changeset, mergeData.fNumDstEntries);
}
 
// 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.
void RNTupleMerger::MergeCommonColumns(RClusterPool &clusterPool, DescriptorId_t clusterId,
                                       std::span<RColumnInfo> commonColumns, RCluster::ColumnSet_t commonColumnSet,
                                       RSealedPageMergeData &sealedPageData, const RNTupleMergeData &mergeData)
{
   assert(commonColumns.size() == commonColumnSet.size());
   if (commonColumns.empty())
      return;
 
   const RCluster *cluster = clusterPool.GetCluster(clusterId, commonColumnSet);
   // we expect the cluster pool to contain the requested set of columns, since they were
   // validated by CompareDescriptorStructures().
   assert(cluster);
 
   const auto &clusterDesc = mergeData.fSrcDescriptor->GetClusterDescriptor(clusterId);
 
   for (const auto &column : commonColumns) {
      const auto &columnId = column.fInputId;
      R__ASSERT(clusterDesc.ContainsColumn(columnId));
 
      const auto &columnDesc = mergeData.fSrcDescriptor->GetColumnDescriptor(columnId);
      const auto srcColElement = RColumnElementBase::Generate(columnDesc.GetType());
      const auto dstColElement = RColumnElementBase::Generate(column.fColumnType);
 
      // Now get the pages for this column in this cluster
      const auto &pages = clusterDesc.GetPageRange(columnId);
 
      RPageStorage::SealedPageSequence_t sealedPages;
      sealedPages.resize(pages.fPageInfos.size());
 
      // Each column range potentially has a distinct compression settings
      const auto colRangeCompressionSettings = clusterDesc.GetColumnRange(columnId).fCompressionSettings;
      const bool needsCompressionChange = mergeData.fMergeOpts.fCompressionSettings != kUnknownCompressionSettings &&
                                          colRangeCompressionSettings != mergeData.fMergeOpts.fCompressionSettings;
 
      if (needsCompressionChange && mergeData.fMergeOpts.fExtraVerbose)
         Info("RNTuple::Merge", "Column %s: changing source compression from %d to %d", column.fColumnName.c_str(),
              colRangeCompressionSettings, mergeData.fMergeOpts.fCompressionSettings);
 
      // If the column range is already uncompressed we don't need to allocate any new buffer, so we don't
      // bother reserving memory for them.
      size_t pageBufferBaseIdx = sealedPageData.fBuffers.size();
      if (colRangeCompressionSettings != 0)
         sealedPageData.fBuffers.resize(sealedPageData.fBuffers.size() + pages.fPageInfos.size());
 
      // Loop over the pages
      std::uint64_t pageIdx = 0;
      for (const auto &pageInfo : pages.fPageInfos) {
         assert(pageIdx < sealedPages.size());
         assert(sealedPageData.fBuffers.size() == 0 || pageIdx < sealedPageData.fBuffers.size());
 
         ROnDiskPage::Key key{columnId, pageIdx};
         auto onDiskPage = cluster->GetOnDiskPage(key);
 
         const auto checksumSize = pageInfo.fHasChecksum * RPageStorage::kNBytesPageChecksum;
         RPageStorage::RSealedPage &sealedPage = sealedPages[pageIdx];
         sealedPage.SetNElements(pageInfo.fNElements);
         sealedPage.SetHasChecksum(pageInfo.fHasChecksum);
         sealedPage.SetBufferSize(pageInfo.fLocator.fBytesOnStorage + 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 (needsCompressionChange) {
            const auto uncompressedSize = srcColElement->GetSize() * sealedPage.GetNElements();
            auto &buffer = sealedPageData.fBuffers[pageBufferBaseIdx + pageIdx];
            buffer = std::make_unique<std::uint8_t[]>(uncompressedSize + checksumSize);
            RChangeCompressionFunc compressTask{
               column.fOutputId, *srcColElement, *dstColElement, mergeData.fMergeOpts,
               sealedPage,       *fPageAlloc,    buffer.get(),
            };
 
            if (fTaskGroup)
               fTaskGroup->Run(compressTask);
            else
               compressTask();
         }
 
         ++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
}
 
// Generates default values for columns that are not present in the current source RNTuple
// but are present in the destination's schema.
static void GenerateExtraDstColumns(size_t nClusterEntries, std::span<RColumnInfo> extraDstColumns,
                                    RSealedPageMergeData &sealedPageData, const RNTupleMergeData &mergeData)
{
   for (const auto &column : extraDstColumns) {
      const auto &columnId = column.fInputId;
      const auto &columnDesc = mergeData.fDstDescriptor.GetColumnDescriptor(columnId);
      const RFieldDescriptor *field = column.fParentField;
 
      // Skip all auxiliary columns
      if (field->GetLogicalColumnIds()[0] != columnId)
         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 != kInvalidDescriptorId;) {
         const RFieldDescriptor &parent = mergeData.fSrcDescriptor->GetFieldDescriptor(parentId);
         if (parent.GetStructure() == ENTupleStructure::kCollection ||
             parent.GetStructure() == 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 == ENTupleStructure::kUnsplit) {
         Fatal(
            "RNTuple::Merge",
            "Destination RNTuple contains an Unsplit field (%s) that is not present in one of the sources. "
            "Creating a default value for an Unsplit field is ill-defined, therefore the merging process will abort.",
            field->GetFieldName().c_str());
         continue;
      }
 
      // NOTE: we cannot have a Record here because it has no associated columns.
      R__ASSERT(structure == ENTupleStructure::kCollection || structure == ENTupleStructure::kVariant ||
                structure == ENTupleStructure::kLeaf);
 
      const auto colElement = RColumnElementBase::Generate(columnDesc.GetType());
      const auto nElements = nClusterEntries * nRepetitions;
      const auto bytesOnStorage = colElement->GetPackedSize(nElements);
      constexpr auto kPageSizeLimit = 256 * 1024;
      // TODO(gparolini): consider coalescing the last page if its size is less than some threshold
      const size_t nPages = bytesOnStorage / kPageSizeLimit + !!(bytesOnStorage % kPageSizeLimit);
      for (size_t i = 0; i < nPages; ++i) {
         const auto pageSize = (i < nPages - 1) ? kPageSizeLimit : bytesOnStorage - kPageSizeLimit * (nPages - 1);
         const auto checksumSize = RPageStorage::kNBytesPageChecksum;
         const auto bufSize = pageSize + checksumSize;
         auto &buffer = sealedPageData.fBuffers.emplace_back(new unsigned char[bufSize]);
 
         RPageStorage::RSealedPage sealedPage{buffer.get(), bufSize, static_cast<std::uint32_t>(nElements), true};
         memset(buffer.get(), 0, pageSize);
         sealedPage.ChecksumIfEnabled();
 
         sealedPageData.fPagesV.push_back({sealedPage});
      }
 
      sealedPageData.fGroups.emplace_back(column.fOutputId, sealedPageData.fPagesV.back().cbegin(),
                                          sealedPageData.fPagesV.back().cend());
   }
}
 
// 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.
void RNTupleMerger::MergeSourceClusters(RPageSource &source, std::span<RColumnInfo> commonColumns,
                                        std::span<RColumnInfo> extraDstColumns, RNTupleMergeData &mergeData)
{
   RClusterPool clusterPool{source};
 
   // Convert columns to a ColumnSet for the ClusterPool query
   RCluster::ColumnSet_t commonColumnSet;
   commonColumnSet.reserve(commonColumns.size());
   for (const auto &column : commonColumns)
      commonColumnSet.emplace(column.fInputId);
 
   RCluster::ColumnSet_t extraDstColumnSet;
   extraDstColumnSet.reserve(extraDstColumns.size());
   for (const auto &column : extraDstColumns)
      extraDstColumnSet.emplace(column.fInputId);
 
   // Loop over all clusters in this file.
   // descriptor->GetClusterIterable() doesn't guarantee any specific order, so we explicitly
   // request the first cluster.
   DescriptorId_t clusterId = mergeData.fSrcDescriptor->FindClusterId(0, 0);
   while (clusterId != kInvalidDescriptorId) {
      const auto &clusterDesc = mergeData.fSrcDescriptor->GetClusterDescriptor(clusterId);
      const auto nClusterEntries = clusterDesc.GetNEntries();
      R__ASSERT(nClusterEntries > 0);
 
      RSealedPageMergeData sealedPageData;
 
      if (!commonColumnSet.empty()) {
         MergeCommonColumns(clusterPool, clusterId, commonColumns, commonColumnSet, sealedPageData, mergeData);
      }
 
      if (!extraDstColumnSet.empty()) {
         GenerateExtraDstColumns(nClusterEntries, extraDstColumns, sealedPageData, mergeData);
      }
 
      // Commit the pages and the clusters
      mergeData.fDestination.CommitSealedPageV(sealedPageData.fGroups);
      mergeData.fDestination.CommitCluster(nClusterEntries);
      mergeData.fNumDstEntries += nClusterEntries;
 
      // Go to the next cluster
      clusterId = mergeData.fSrcDescriptor->FindNextClusterId(clusterId);
   }
 
   // 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.
}
 
// Given a field, fill `columns` and `colIdMap` with information about all columns belonging to it and its subfields.
// `colIdMap` 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 `CompareDescriptorStructures()` earlier, we should be guaranteed that two matching
// columns will have at least compatible representations.
static void AddColumnsFromField(std::vector<RColumnInfo> &columns, const RNTupleDescriptor &srcDesc,
                                RNTupleMergeData &mergeData, const RFieldDescriptor &fieldDesc,
                                const std::string &prefix = "")
{
   std::string name = prefix + '.' + fieldDesc.GetFieldName();
 
   const auto &columnIds = fieldDesc.GetLogicalColumnIds();
   columns.reserve(columns.size() + columnIds.size());
   for (const auto &columnId : columnIds) {
      const auto &srcColumn = srcDesc.GetColumnDescriptor(columnId);
      RColumnInfo info;
      info.fColumnName = name + '.' + std::to_string(srcColumn.GetIndex());
      info.fInputId = columnId;
      info.fParentField = &fieldDesc;
 
      if (auto it = mergeData.fColumnIdMap.find(info.fColumnName); it != mergeData.fColumnIdMap.end()) {
         info.fOutputId = it->second.fColumnId;
         info.fColumnType = it->second.fColumnType;
      } else {
         info.fOutputId = mergeData.fColumnIdMap.size();
         // NOTE(gparolini): map the type of src column to the type of dst column.
         // This mapping is only relevant for common columns and it's done to ensure we keep a consistent
         // on-disk representation of the same column.
         // This is also important to do for first source when it is used to generate the destination sink,
         // because even in that case their column representations may differ.
         // e.g. if the destination has a different compression than the source, an integer column might be
         // zigzag-encoded in the source but not in the destination.
         const auto &dstColumn = (&mergeData.fDstDescriptor == &srcDesc)
                                    ? srcColumn
                                    : mergeData.fDstDescriptor.GetColumnDescriptor(columnId);
         info.fColumnType = dstColumn.GetType();
         mergeData.fColumnIdMap[info.fColumnName] = {info.fOutputId, info.fColumnType};
      }
      columns.emplace_back(info);
   }
 
   for (const auto &field : srcDesc.GetFieldIterable(fieldDesc))
      AddColumnsFromField(columns, srcDesc, mergeData, field, name);
}
 
// Converts the fields comparison data to the corresponding column information.
// While doing so, it collects such information in `colIdMap`, which is used by later calls to this function
// to map already-seen column names to their chosen outputId, type and so on.
static RColumnInfoGroup
GatherColumnInfos(const RDescriptorsComparison &descCmp, const RNTupleDescriptor &srcDesc, RNTupleMergeData &mergeData)
{
   RColumnInfoGroup res;
   for (const RFieldDescriptor *field : descCmp.fExtraDstFields) {
      AddColumnsFromField(res.fExtraDstColumns, mergeData.fDstDescriptor, mergeData, *field);
   }
   for (const auto *field : descCmp.fCommonFields) {
      AddColumnsFromField(res.fCommonColumns, srcDesc, mergeData, *field);
   }
   return res;
}
 
RNTupleMerger::RNTupleMerger()
   // TODO(gparolini): consider using an arena allocator instead, since we know the precise lifetime
   // of the RNTuples we are going to handle (e.g. we can reset the arena at every source)
   : fPageAlloc(std::make_unique<RPageAllocatorHeap>())
{
#ifdef R__USE_IMT
   if (ROOT::IsImplicitMTEnabled())
      fTaskGroup = TTaskGroup();
#endif
}
 
RResult<void>
RNTupleMerger::Merge(std::span<RPageSource *> sources, RPageSink &destination, const RNTupleMergeOptions &mergeOpts)
{
   RNTupleMergeData mergeData{sources, destination, mergeOpts};
 
   std::unique_ptr<RNTupleModel> model; // used to initialize the schema of the output RNTuple
 
#define SKIP_OR_ABORT(errMsg)                                                        \
   do {                                                                              \
      if (mergeOpts.fErrBehavior == ENTupleMergeErrBehavior::kSkip) {                \
         Warning("RNTuple::Merge", "Skipping RNTuple due to: %s", (errMsg).c_str()); \
         continue;                                                                   \
      } else {                                                                       \
         return R__FAIL(errMsg);                                                     \
      }                                                                              \
   } while (0)
 
   // Merge main loop
   for (RPageSource *source : sources) {
      source->Attach();
      auto srcDescriptor = source->GetSharedDescriptorGuard();
      mergeData.fSrcDescriptor = &srcDescriptor.GetRef();
 
      // Create sink from the input model if not initialized
      if (!destination.IsInitialized()) {
         model = srcDescriptor->CreateModel();
         destination.Init(*model);
      }
 
      for (const auto &extraTypeInfoDesc : srcDescriptor->GetExtraTypeInfoIterable())
         destination.UpdateExtraTypeInfo(extraTypeInfoDesc);
 
      auto descCmpRes = CompareDescriptorStructure(mergeData.fDstDescriptor, srcDescriptor.GetRef());
      if (!descCmpRes) {
         SKIP_OR_ABORT(
            std::string("Source RNTuple will be skipped due to incompatible schema with the destination:\n") +
            descCmpRes.GetError()->GetReport());
      }
      auto descCmp = descCmpRes.Unwrap();
 
      // 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.size()) {
         if (mergeOpts.fMergingMode == ENTupleMergingMode::kUnion) {
            // late model extension for all fExtraSrcFields in Union mode
            ExtendDestinationModel(descCmp.fExtraSrcFields, *model, mergeData);
            descCmp.fCommonFields.insert(descCmp.fCommonFields.end(), descCmp.fExtraSrcFields.begin(),
                                         descCmp.fExtraSrcFields.end());
         } 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);
         }
      }
 
      // handle extra dst fields & common fields
      auto columnInfos = GatherColumnInfos(descCmp, srcDescriptor.GetRef(), mergeData);
      MergeSourceClusters(*source, columnInfos.fCommonColumns, columnInfos.fExtraDstColumns, mergeData);
   } // end loop over sources
 
   // Commit the output
   destination.CommitClusterGroup();
   destination.CommitDataset();
 
   return RResult<void>::Success();
}