[ntuple] Add Merger tests for Streamer fields

Also change the failure in GenerateZeroPages from a R__LOG_FATAL to a
ROOT::Result failure

Author: silverweed

tree/ntuple/inc/ROOT/RNTupleMerger.hxx

@@ -33,7 +33,7 @@ class RNTuple;
 namespace Internal {
 class RPageAllocator;
 class RClusterPool;
-}
+} // namespace Internal
 
 namespace Experimental::Internal {
 
@@ -49,6 +49,12 @@ enum class ENTupleMergingMode {
    kUnion
 };
 
+inline const char *ToString(ENTupleMergingMode mode)
+{
+   static const char *const kMergingModeStr[] = {"Filter", "Strict", "Union"};
+   return kMergingModeStr[static_cast<int>(mode)];
+}
+
 enum class ENTupleMergeErrBehavior {
    /// The merger will abort merging as soon as an error is encountered
    kAbort,
@@ -98,14 +104,18 @@ class RNTupleMerger final {
    std::optional<TTaskGroup> fTaskGroup;
    std::unique_ptr<ROOT::RNTupleModel> fModel;
 
-   void MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool, const ROOT::RClusterDescriptor &clusterDesc,
-                           std::span<const RColumnMergeInfo> commonColumns,
-                           const ROOT::Internal::RCluster::ColumnSet_t &commonColumnSet,
-                           std::size_t nCommonColumnsInCluster, RSealedPageMergeData &sealedPageData,
-                           const RNTupleMergeData &mergeData, ROOT::Internal::RPageAllocator &pageAlloc);
-
-   void MergeSourceClusters(ROOT::Internal::RPageSource &source, std::span<const RColumnMergeInfo> commonColumns,
-                            std::span<const RColumnMergeInfo> extraDstColumns, RNTupleMergeData &mergeData);
+   [[nodiscard]]
+   ROOT::RResult<void> MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool,
+                                          const ROOT::RClusterDescriptor &clusterDesc,
+                                          std::span<const RColumnMergeInfo> commonColumns,
+                                          const ROOT::Internal::RCluster::ColumnSet_t &commonColumnSet,
+                                          std::size_t nCommonColumnsInCluster, RSealedPageMergeData &sealedPageData,
+                                          const RNTupleMergeData &mergeData, ROOT::Internal::RPageAllocator &pageAlloc);
+
+   [[nodiscard]]
+   ROOT::RResult<void>
+   MergeSourceClusters(ROOT::Internal::RPageSource &source, std::span<const RColumnMergeInfo> commonColumns,
+                       std::span<const RColumnMergeInfo> extraDstColumns, RNTupleMergeData &mergeData);
 
    /// Creates a RNTupleMerger with the given destination.
    /// The model must be given if and only if `destination` has been initialized with that model

tree/ntuple/src/RNTupleMerger.cxx

@@ -647,12 +647,14 @@ static void ExtendDestinationModel(std::span<const ROOT::RFieldDescriptor *> new
 }
 
 // Generates default (zero) values for the given columns
-static void GenerateZeroPagesForColumns(size_t nEntriesToGenerate, std::span<const RColumnMergeInfo> columns,
-                                        RSealedPageMergeData &sealedPageData, ROOT::Internal::RPageAllocator &pageAlloc,
-                                        const ROOT::RNTupleDescriptor &dstDescriptor, const RNTupleMergeData &mergeData)
+[[nodiscard]]
+static ROOT::RResult<void>
+GenerateZeroPagesForColumns(size_t nEntriesToGenerate, std::span<const RColumnMergeInfo> columns,
+                            RSealedPageMergeData &sealedPageData, ROOT::Internal::RPageAllocator &pageAlloc,
+                            const ROOT::RNTupleDescriptor &dstDescriptor, const RNTupleMergeData &mergeData)
 {
    if (!nEntriesToGenerate)
-      return;
+      return ROOT::RResult<void>::Success();
 
    for (const auto &column : columns) {
       const ROOT::RFieldDescriptor *field = column.fParentFieldDescriptor;
@@ -682,12 +684,10 @@ static void GenerateZeroPagesForColumns(size_t nEntriesToGenerate, std::span<con
       const auto structure = field->GetStructure();
 
       if (structure == ROOT::ENTupleStructure::kStreamer) {
-         R__LOG_FATAL(NTupleMergeLog())
-            << "RNTuple::Merge"
-               "Destination RNTuple contains a streamer field (%s) 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."
-            << field->GetFieldName();
-         continue;
+         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.
@@ -726,21 +726,23 @@ static void GenerateZeroPagesForColumns(size_t nEntriesToGenerate, std::span<con
                                              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.
-void RNTupleMerger::MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool,
-                                       const ROOT::RClusterDescriptor &clusterDesc,
-                                       std::span<const RColumnMergeInfo> commonColumns,
-                                       const RCluster::ColumnSet_t &commonColumnSet,
-                                       std::size_t nCommonColumnsInCluster, RSealedPageMergeData &sealedPageData,
-                                       const RNTupleMergeData &mergeData, ROOT::Internal::RPageAllocator &pageAlloc)
+ROOT::RResult<void>
+RNTupleMerger::MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool,
+                                  const ROOT::RClusterDescriptor &clusterDesc,
+                                  std::span<const RColumnMergeInfo> commonColumns,
+                                  const RCluster::ColumnSet_t &commonColumnSet, std::size_t nCommonColumnsInCluster,
+                                  RSealedPageMergeData &sealedPageData, const RNTupleMergeData &mergeData,
+                                  ROOT::Internal::RPageAllocator &pageAlloc)
 {
    assert(nCommonColumnsInCluster == commonColumnSet.size());
    assert(nCommonColumnsInCluster <= commonColumns.size());
    if (nCommonColumnsInCluster == 0)
-      return;
+      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
@@ -795,8 +797,10 @@ void RNTupleMerger::MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool
       // TODO: also avoid doing this if we added no real page of this column to the destination yet.
       if (columnDesc.GetFirstElementIndex() > clusterDesc.GetFirstEntryIndex() && mergeData.fNumDstEntries > 0) {
          const auto nMissingEntries = columnDesc.GetFirstElementIndex() - clusterDesc.GetFirstEntryIndex();
-         GenerateZeroPagesForColumns(nMissingEntries, {&column, 1}, sealedPageData, pageAlloc, mergeData.fDstDescriptor,
-                                     mergeData);
+         auto res = GenerateZeroPagesForColumns(nMissingEntries, {&column, 1}, sealedPageData, pageAlloc,
+                                                mergeData.fDstDescriptor, mergeData);
+         if (!res)
+            return R__FORWARD_ERROR(res);
       }
 
       // Loop over the pages
@@ -844,15 +848,18 @@ void RNTupleMerger::MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool
       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.
-void RNTupleMerger::MergeSourceClusters(RPageSource &source, std::span<const RColumnMergeInfo> commonColumns,
-                                        std::span<const RColumnMergeInfo> extraDstColumns, RNTupleMergeData &mergeData)
+ROOT::RResult<void>
+RNTupleMerger::MergeSourceClusters(RPageSource &source, std::span<const RColumnMergeInfo> commonColumns,
+                                   std::span<const RColumnMergeInfo> extraDstColumns, RNTupleMergeData &mergeData)
 {
    ROOT::Internal::RClusterPool clusterPool{source};
 
@@ -894,10 +901,15 @@ void RNTupleMerger::MergeSourceClusters(RPageSource &source, std::span<const RCo
          missingColumns.push_back(commonColumns[i]);
 
       RSealedPageMergeData sealedPageData;
-      MergeCommonColumns(clusterPool, clusterDesc, commonColumns, commonColumnSet, nCommonColumnsInCluster,
-                         sealedPageData, mergeData, *fPageAlloc);
-      GenerateZeroPagesForColumns(nClusterEntries, missingColumns, sealedPageData, *fPageAlloc,
-                                  mergeData.fDstDescriptor, mergeData);
+      auto res = MergeCommonColumns(clusterPool, clusterDesc, commonColumns, commonColumnSet, nCommonColumnsInCluster,
+                                    sealedPageData, mergeData, *fPageAlloc);
+      if (!res)
+         return R__FORWARD_ERROR(res);
+
+      res = GenerateZeroPagesForColumns(nClusterEntries, missingColumns, sealedPageData, *fPageAlloc,
+                                        mergeData.fDstDescriptor, mergeData);
+      if (!res)
+         return R__FORWARD_ERROR(res);
 
       // Commit the pages and the clusters
       mergeData.fDestination.CommitSealedPageV(sealedPageData.fGroups);
@@ -914,6 +926,7 @@ void RNTupleMerger::MergeSourceClusters(RPageSource &source, std::span<const RCo
    // 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();
 }
 
 static std::optional<std::type_index> ColumnInMemoryType(std::string_view fieldType, ENTupleColumnType onDiskType)
@@ -977,7 +990,8 @@ static void AddColumnsFromField(std::vector<RColumnMergeInfo> &columns, const RO
       // 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()).
+      // 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
@@ -1183,7 +1197,9 @@ ROOT::RResult<void> RNTupleMerger::Merge(std::span<RPageSource *> sources, const
 
       // handle extra dst fields & common fields
       auto columnInfos = GatherColumnInfos(descCmp, srcDescriptor.GetRef(), mergeData);
-      MergeSourceClusters(*source, columnInfos.fCommonColumns, columnInfos.fExtraDstColumns, mergeData);
+      auto res = MergeSourceClusters(*source, columnInfos.fCommonColumns, columnInfos.fExtraDstColumns, mergeData);
+      if (!res)
+         return R__FORWARD_ERROR(res);
    } // end loop over sources
 
    if (fDestination->GetNEntries() == 0)