Logo ROOT  
Reference Guide
 
Loading...
Searching...
No Matches
RNTupleMerger.cxx
Go to the documentation of this file.
1/// \file RNTupleMerger.cxx
2/// \ingroup NTuple
3/// \author Jakob Blomer <jblomer@cern.ch>, Max Orok <maxwellorok@gmail.com>, Alaettin Serhan Mete <amete@anl.gov>,
4/// Giacomo Parolini <giacomo.parolini@cern.ch>
5/// \date 2020-07-08
6/// \warning This is part of the ROOT 7 prototype! It will
7/// change without notice. It might trigger earthquakes. Feedback is welcome!
8
9/*************************************************************************
10 * Copyright (C) 1995-2020, Rene Brun and Fons Rademakers. *
11 * All rights reserved. *
12 * *
13 * For the licensing terms see $ROOTSYS/LICENSE. *
14 * For the list of contributors see $ROOTSYS/README/CREDITS. *
15 *************************************************************************/
16
17#include <ROOT/RError.hxx>
18#include <ROOT/RNTuple.hxx>
21#include <ROOT/RNTupleModel.hxx>
22#include <ROOT/RNTupleTypes.hxx>
23#include <ROOT/RNTupleUtils.hxx>
26#include <ROOT/RPageStorage.hxx>
27#include <ROOT/RClusterPool.hxx>
29#include <ROOT/RNTupleZip.hxx>
31#include <TROOT.h>
32#include <TFileMergeInfo.h>
33#include <TFile.h>
34#include <TKey.h>
35
36#include <algorithm>
37#include <deque>
38#include <inttypes.h> // for PRIu64
39#include <initializer_list>
40#include <unordered_map>
41#include <vector>
42
53
54using namespace ROOT::Experimental::Internal;
55
57{
58 static ROOT::RLogChannel sLog("ROOT.NTuple.Merge");
59 return sLog;
60}
61
62// TFile options parsing
63// -------------------------------------------------------------------------------------
64static bool BeginsWithDelimitedWord(const TString &str, const char *word)
65{
66 const Ssiz_t wordLen = strlen(word);
67 if (str.Length() < wordLen)
68 return false;
69 if (!str.BeginsWith(word, TString::ECaseCompare::kIgnoreCase))
70 return false;
71 return str.Length() == wordLen || str(wordLen) == ' ';
72}
73
74template <typename T>
75static std::optional<T> ParseStringOption(const TString &opts, const char *pattern,
76 std::initializer_list<std::pair<const char *, T>> validValues)
77{
78 const Ssiz_t patternLen = strlen(pattern);
79 assert(pattern[patternLen - 1] == '='); // we want to parse options with the format `option=Value`
80 if (auto idx = opts.Index(pattern, 0, TString::ECaseCompare::kIgnoreCase);
81 idx >= 0 && opts.Length() > idx + patternLen) {
82 auto sub = TString(opts(idx + patternLen, opts.Length() - idx - patternLen));
83 for (const auto &[name, value] : validValues) {
84 if (BeginsWithDelimitedWord(sub, name)) {
85 return value;
86 }
87 }
88 }
89 return std::nullopt;
90}
91
92static std::optional<ENTupleMergingMode> ParseOptionMergingMode(const TString &opts)
93{
94 return ParseStringOption<ENTupleMergingMode>(opts, "rntuple.MergingMode=",
95 {
96 {"Filter", ENTupleMergingMode::kFilter},
97 {"Union", ENTupleMergingMode::kUnion},
98 {"Strict", ENTupleMergingMode::kStrict},
99 });
100}
101
102static std::optional<ENTupleMergeErrBehavior> ParseOptionErrBehavior(const TString &opts)
103{
104 return ParseStringOption<ENTupleMergeErrBehavior>(opts, "rntuple.ErrBehavior=",
105 {
106 {"Abort", ENTupleMergeErrBehavior::kAbort},
107 {"Skip", ENTupleMergeErrBehavior::kSkip},
108 });
109}
110// -------------------------------------------------------------------------------------
111
112// Entry point for TFileMerger. Internally calls RNTupleMerger::Merge().
114// IMPORTANT: this function must not throw, as it is used in exception-unsafe code (TFileMerger).
115try {
116 // Check the inputs
117 if (!inputs || inputs->GetEntries() < 3 || !mergeInfo) {
118 R__LOG_ERROR(NTupleMergeLog()) << "Invalid inputs.";
119 return -1;
120 }
121
122 // Parse the input parameters
124
125 // First entry is the RNTuple name
126 std::string ntupleName = std::string(itr()->GetName());
127
128 // Second entry is the output file
129 TObject *secondArg = itr();
130 TFile *outFile = dynamic_cast<TFile *>(secondArg);
131 if (!outFile) {
132 R__LOG_ERROR(NTupleMergeLog()) << "Second input parameter should be a TFile, but it's a "
133 << secondArg->ClassName() << ".";
134 return -1;
135 }
136
137 // Check if the output file already has a key with that name
138 TKey *outKey = outFile->FindKey(ntupleName.c_str());
139 ROOT::RNTuple *outNTuple = nullptr;
140 if (outKey) {
141 outNTuple = outKey->ReadObject<ROOT::RNTuple>();
142 if (!outNTuple) {
143 R__LOG_ERROR(NTupleMergeLog()) << "Output file already has key, but not of type RNTuple!";
144 return -1;
145 }
146 // In principle, we should already be working on the RNTuple object from the output file, but just continue with
147 // pointer we just got.
148 }
149
150 const bool defaultComp = mergeInfo->fOptions.Contains("DefaultCompression");
151 const bool firstSrcComp = mergeInfo->fOptions.Contains("FirstSrcCompression");
152 const bool extraVerbose = mergeInfo->fOptions.Contains("rntuple.ExtraVerbose");
153 if (defaultComp && firstSrcComp) {
154 // this should never happen through hadd, but a user may call RNTuple::Merge() from custom code.
155 R__LOG_WARNING(NTupleMergeLog()) << "Passed both options \"DefaultCompression\" and \"FirstSrcCompression\": "
156 "only the latter will apply.";
157 }
158 std::optional<std::uint32_t> compression;
159 if (firstSrcComp) {
160 // user passed -ff or -fk: use the same compression as the first RNTuple we find in the sources.
161 // (do nothing here, the compression will be fetched below)
162 } else if (!defaultComp) {
163 // compression was explicitly passed by the user: use it.
164 compression = outFile->GetCompressionSettings();
165 } else {
166 // user passed no compression-related options: use default
168 R__LOG_INFO(NTupleMergeLog()) << "Using the default compression: " << *compression;
169 }
170
171 // The remaining entries are the input files
172 std::vector<std::unique_ptr<RPageSourceFile>> sources;
173 std::vector<RPageSource *> sourcePtrs;
174
175 while (const auto &pitr = itr()) {
176 TFile *inFile = dynamic_cast<TFile *>(pitr);
177 ROOT::RNTuple *anchor = inFile ? inFile->Get<ROOT::RNTuple>(ntupleName.c_str()) : nullptr;
178 if (!anchor) {
179 R__LOG_INFO(NTupleMergeLog()) << "No RNTuple anchor named '" << ntupleName << "' from file '"
180 << inFile->GetName() << "'";
181 continue;
182 }
183
185 if (!compression) {
186 // Get the compression of this RNTuple and use it as the output compression.
187 // We currently assume all column ranges have the same compression, so we just peek at the first one.
188 source->Attach(RNTupleSerializer::EDescriptorDeserializeMode::kRaw);
189 auto descriptor = source->GetSharedDescriptorGuard();
190 auto clusterIter = descriptor->GetClusterIterable();
192 if (firstCluster == clusterIter.end()) {
194 << "Asked to use the first source's compression as the output compression, but the "
195 "first source (file '"
196 << inFile->GetName()
197 << "') has an empty RNTuple, therefore the output compression could not be "
198 "determined.";
199 return -1;
200 }
201 auto colRangeIter = (*firstCluster).GetColumnRangeIterable();
203 if (firstColRange == colRangeIter.end()) {
205 << "Asked to use the first source's compression as the output compression, but the "
206 "first source (file '"
207 << inFile->GetName()
208 << "') has an empty RNTuple, therefore the output compression could not be "
209 "determined.";
210 return -1;
211 }
212 compression = (*firstColRange).GetCompressionSettings().value();
213 R__LOG_INFO(NTupleMergeLog()) << "Using the first RNTuple's compression: " << *compression;
214 }
215 sources.push_back(std::move(source));
216 }
217
220 writeOpts.SetCompression(*compression);
221 auto destination = std::make_unique<ROOT::Internal::RPageSinkFile>(ntupleName, *outFile, writeOpts);
222 std::unique_ptr<ROOT::RNTupleModel> model;
223 // If we already have an existing RNTuple, copy over its descriptor to support incremental merging
224 if (outNTuple) {
226 outSource->Attach(RNTupleSerializer::EDescriptorDeserializeMode::kForWriting);
227 auto desc = outSource->GetSharedDescriptorGuard();
228 model = destination->InitFromDescriptor(desc.GetRef(), true /* copyClusters */);
229 }
230
231 // Interface conversion
232 sourcePtrs.reserve(sources.size());
233 for (const auto &s : sources) {
234 sourcePtrs.push_back(s.get());
235 }
236
237 // Now merge
238 RNTupleMerger merger{std::move(destination), std::move(model)};
240 mergerOpts.fCompressionSettings = *compression;
241 mergerOpts.fExtraVerbose = extraVerbose;
242 if (auto mergingMode = ParseOptionMergingMode(mergeInfo->fOptions)) {
243 mergerOpts.fMergingMode = *mergingMode;
244 }
245 if (auto errBehavior = ParseOptionErrBehavior(mergeInfo->fOptions)) {
246 mergerOpts.fErrBehavior = *errBehavior;
247 }
248 merger.Merge(sourcePtrs, mergerOpts).ThrowOnError();
249
250 // Provide the caller with a merged anchor object (even though we've already
251 // written it).
252 *this = *outFile->Get<ROOT::RNTuple>(ntupleName.c_str());
253
254 return 0;
255} catch (const std::exception &ex) {
256 R__LOG_ERROR(NTupleMergeLog()) << "Exception thrown while merging: " << ex.what();
257 return -1;
258}
259
260namespace {
261// Functor used to change the compression of a page to `options.fCompressionSettings`.
262struct RChangeCompressionFunc {
263 const RColumnElementBase &fSrcColElement;
264 const RColumnElementBase &fDstColElement;
265 const RNTupleMergeOptions &fMergeOptions;
266
267 RPageStorage::RSealedPage &fSealedPage;
269 std::uint8_t *fBuffer;
270
271 void operator()() const
272 {
273 auto page = RPageSource::UnsealPage(fSealedPage, fSrcColElement, fPageAlloc).Unwrap();
275 sealConf.fElement = &fDstColElement;
276 sealConf.fPage = &page;
277 sealConf.fBuffer = fBuffer;
278 sealConf.fCompressionSettings = *fMergeOptions.fCompressionSettings;
279 sealConf.fWriteChecksum = fSealedPage.GetHasChecksum();
280 auto refSealedPage = RPageSink::SealPage(sealConf);
281 fSealedPage = refSealedPage;
282 }
283};
284
285struct RCommonField {
286 const ROOT::RFieldDescriptor *fSrc;
287 const ROOT::RFieldDescriptor *fDst;
288
289 RCommonField(const ROOT::RFieldDescriptor &src, const ROOT::RFieldDescriptor &dst) : fSrc(&src), fDst(&dst) {}
290};
291
292struct RDescriptorsComparison {
293 std::vector<const ROOT::RFieldDescriptor *> fExtraDstFields;
294 std::vector<const ROOT::RFieldDescriptor *> fExtraSrcFields;
295 std::vector<RCommonField> fCommonFields;
296};
297
298struct RColumnOutInfo {
299 ROOT::DescriptorId_t fColumnId;
300 ENTupleColumnType fColumnType;
301};
302
303// { fully.qualified.fieldName.colInputId => colOutputInfo }
304using ColumnIdMap_t = std::unordered_map<std::string, RColumnOutInfo>;
305
306struct RColumnInfoGroup {
307 std::vector<RColumnMergeInfo> fExtraDstColumns;
308 // These are sorted by InputId
309 std::vector<RColumnMergeInfo> fCommonColumns;
310};
311
312} // namespace
313
314// These structs cannot be in the anon namespace becase they're used in RNTupleMerger's private interface.
317 // This column name is built as a dot-separated concatenation of the ancestry of
318 // the columns' parent fields' names plus the index of the column itself.
319 // e.g. "Muon.pt.x._0"
320 std::string fColumnName;
321 // The column id in the source RNTuple
323 // The corresponding column id in the destination RNTuple (the mapping happens in AddColumnsFromField())
326 // If nullopt, use the default in-memory type
327 std::optional<std::type_index> fInMemoryType;
330};
331
332// Data related to a single call of RNTupleMerger::Merge()
350
352 // We use a std::deque so that references to the contained SealedPageSequence_t, and its iterators, are
353 // never invalidated.
354 std::deque<RPageStorage::SealedPageSequence_t> fPagesV;
355 std::vector<RPageStorage::RSealedPageGroup> fGroups;
356 std::vector<std::unique_ptr<std::uint8_t[]>> fBuffers;
357};
358
359std::ostream &operator<<(std::ostream &os, const std::optional<ROOT::RColumnDescriptor::RValueRange> &x)
360{
361 if (x) {
362 os << '(' << x->fMin << ", " << x->fMax << ')';
363 } else {
364 os << "(null)";
365 }
366 return os;
367}
368
369} // namespace ROOT::Experimental::Internal
370
372{
373 // clang-format off
394 // clang-format on
395 return false;
396}
397
398/// Compares the top level fields of `dst` and `src` and determines whether they can be merged or not.
399/// In addition, returns the differences between `dst` and `src`'s structures
402{
403 // Cases:
404 // 1. dst == src
405 // 2. dst has fields that src hasn't
406 // 3. src has fields that dst hasn't
407 // 4. dst and src have fields that differ (compatible or incompatible)
408
409 std::vector<std::string> errors;
410 RDescriptorsComparison res;
411
412 std::vector<RCommonField> commonFields;
413
414 for (const auto &dstField : dst.GetTopLevelFields()) {
415 const auto srcFieldId = src.FindFieldId(dstField.GetFieldName());
417 const auto &srcField = src.GetFieldDescriptor(srcFieldId);
418 commonFields.push_back({srcField, dstField});
419 } else {
420 res.fExtraDstFields.emplace_back(&dstField);
421 }
422 }
423 for (const auto &srcField : src.GetTopLevelFields()) {
424 const auto dstFieldId = dst.FindFieldId(srcField.GetFieldName());
426 res.fExtraSrcFields.push_back(&srcField);
427 }
428
429 // Check compatibility of common fields
431 // NOTE: using index-based for loop because the collection may get extended by the iteration
432 for (std::size_t fieldIdx = 0; fieldIdx < fieldsToCheck.size(); ++fieldIdx) {
433 const auto &field = fieldsToCheck[fieldIdx];
434
435 // NOTE: field.fSrc and field.fDst have the same name by construction
436 const auto &fieldName = field.fSrc->GetFieldName();
437
438 // Require that fields are both projected or both not projected
439 bool projCompatible = field.fSrc->IsProjectedField() == field.fDst->IsProjectedField();
440 if (!projCompatible) {
441 std::stringstream ss;
442 ss << "Field `" << fieldName << "` is incompatible with previously-seen field with that name because the "
443 << (field.fSrc->IsProjectedField() ? "new" : "old") << " one is projected and the other isn't";
444 errors.push_back(ss.str());
445 } else if (field.fSrc->IsProjectedField()) {
446 // if both fields are projected, verify that they point to the same real field
447 const auto srcName = src.GetQualifiedFieldName(field.fSrc->GetProjectionSourceId());
448 const auto dstName = dst.GetQualifiedFieldName(field.fDst->GetProjectionSourceId());
449 if (srcName != dstName) {
450 std::stringstream ss;
451 ss << "Field `" << fieldName
452 << "` is projected to a different field than a previously-seen field with the same name (old: "
453 << dstName << ", new: " << srcName << ")";
454 errors.push_back(ss.str());
455 }
456 }
457
458 // Require that fields types match
459 // TODO(gparolini): allow non-identical but compatible types
460 const auto &srcTyName = field.fSrc->GetTypeName();
461 const auto &dstTyName = field.fDst->GetTypeName();
462 if (srcTyName != dstTyName) {
463 std::stringstream ss;
464 ss << "Field `" << fieldName
465 << "` has a type incompatible with a previously-seen field with the same name: (old: " << dstTyName
466 << ", new: " << srcTyName << ")";
467 errors.push_back(ss.str());
468 }
469
470 // Require that type checksums match
471 const auto srcTyChk = field.fSrc->GetTypeChecksum();
472 const auto dstTyChk = field.fDst->GetTypeChecksum();
473 if (srcTyChk && dstTyChk && *srcTyChk != *dstTyChk) {
474 std::stringstream ss;
475 ss << "Field `" << field.fSrc->GetFieldName()
476 << "` has a different type checksum than previously-seen field with the same name";
477 errors.push_back(ss.str());
478 }
479
480 // Require that type versions match
481 const auto srcTyVer = field.fSrc->GetTypeVersion();
482 const auto dstTyVer = field.fDst->GetTypeVersion();
483 if (srcTyVer != dstTyVer) {
484 std::stringstream ss;
485 ss << "Field `" << field.fSrc->GetFieldName()
486 << "` has a different type version than previously-seen field with the same name (old: " << dstTyVer
487 << ", new: " << srcTyVer << ")";
488 errors.push_back(ss.str());
489 }
490
491 const auto srcRole = field.fSrc->GetStructure();
492 const auto dstRole = field.fDst->GetStructure();
493 if (srcRole != dstRole) {
494 std::stringstream ss;
495 ss << "Field `" << field.fSrc->GetFieldName()
496 << "` has a different structural role than previously-seen field with the same name (old: " << dstRole
497 << ", new: " << srcRole << ")";
498 errors.push_back(ss.str());
499 }
500
501 // Require that column representations match
502 const auto srcNCols = field.fSrc->GetLogicalColumnIds().size();
503 const auto dstNCols = field.fDst->GetLogicalColumnIds().size();
504 if (srcNCols != dstNCols) {
505 std::stringstream ss;
506 ss << "Field `" << field.fSrc->GetFieldName()
507 << "` has a different number of columns than previously-seen field with the same name (old: " << dstNCols
508 << ", new: " << srcNCols << ")";
509 errors.push_back(ss.str());
510 } else {
511 for (auto i = 0u; i < srcNCols; ++i) {
512 const auto srcColId = field.fSrc->GetLogicalColumnIds()[i];
513 const auto dstColId = field.fDst->GetLogicalColumnIds()[i];
514 const auto &srcCol = src.GetColumnDescriptor(srcColId);
515 const auto &dstCol = dst.GetColumnDescriptor(dstColId);
516 // TODO(gparolini): currently we refuse to merge columns of different types unless they are Split/non-Split
517 // version of the same type, because we know how to treat that specific case. We should also properly handle
518 // different but compatible types.
519 if (srcCol.GetType() != dstCol.GetType() &&
520 !IsSplitOrUnsplitVersionOf(srcCol.GetType(), dstCol.GetType())) {
521 std::stringstream ss;
522 ss << i << "-th column of field `" << field.fSrc->GetFieldName()
523 << "` has a different column type of the same column on the previously-seen field with the same name "
524 "(old: "
526 << ", new: " << RColumnElementBase::GetColumnTypeName(dstCol.GetType()) << ")";
527 errors.push_back(ss.str());
528 }
529 if (srcCol.GetBitsOnStorage() != dstCol.GetBitsOnStorage()) {
530 std::stringstream ss;
531 ss << i << "-th column of field `" << field.fSrc->GetFieldName()
532 << "` has a different number of bits of the same column on the previously-seen field with the same "
533 "name "
534 "(old: "
535 << srcCol.GetBitsOnStorage() << ", new: " << dstCol.GetBitsOnStorage() << ")";
536 errors.push_back(ss.str());
537 }
538 if (srcCol.GetValueRange() != dstCol.GetValueRange()) {
539 std::stringstream ss;
540 ss << i << "-th column of field `" << field.fSrc->GetFieldName()
541 << "` has a different value range of the same column on the previously-seen field with the same name "
542 "(old: "
543 << srcCol.GetValueRange() << ", new: " << dstCol.GetValueRange() << ")";
544 errors.push_back(ss.str());
545 }
546 if (srcCol.GetRepresentationIndex() > 0) {
547 std::stringstream ss;
548 ss << i << "-th column of field `" << field.fSrc->GetFieldName()
549 << "` has a representation index higher than 0. This is not supported yet by the merger.";
550 errors.push_back(ss.str());
551 }
552 }
553 }
554
555 // Require that subfields are compatible
556 const auto &srcLinks = field.fSrc->GetLinkIds();
557 const auto &dstLinks = field.fDst->GetLinkIds();
558 if (srcLinks.size() != dstLinks.size()) {
559 std::stringstream ss;
560 ss << "Field `" << field.fSrc->GetFieldName()
561 << "` has a different number of children than previously-seen field with the same name (old: "
562 << dstLinks.size() << ", new: " << srcLinks.size() << ")";
563 errors.push_back(ss.str());
564 } else {
565 for (std::size_t linkIdx = 0, linkNum = srcLinks.size(); linkIdx < linkNum; ++linkIdx) {
566 const auto &srcSubfield = src.GetFieldDescriptor(srcLinks[linkIdx]);
567 const auto &dstSubfield = dst.GetFieldDescriptor(dstLinks[linkIdx]);
568 fieldsToCheck.push_back(RCommonField{srcSubfield, dstSubfield});
569 }
570 }
571 }
572
573 std::string errMsg;
574 for (const auto &err : errors)
575 errMsg += std::string("\n * ") + err;
576
577 if (!errMsg.empty())
578 errMsg = errMsg.substr(1); // strip initial newline
579
580 if (errMsg.length())
581 return R__FAIL(errMsg);
582
583 res.fCommonFields = std::move(commonFields);
584
585 return ROOT::RResult(res);
586}
587
588// Applies late model extension to `destination`, adding all `newFields` to it.
589[[nodiscard]]
591ExtendDestinationModel(std::span<const ROOT::RFieldDescriptor *> newFields, ROOT::RNTupleModel &dstModel,
592 RNTupleMergeData &mergeData, std::vector<RCommonField> &commonFields)
593{
594 assert(newFields.size() > 0); // no point in calling this with 0 new cols
595
596 dstModel.Unfreeze();
598
599 if (mergeData.fMergeOpts.fExtraVerbose) {
600 std::string msg = "destination doesn't contain field";
601 if (newFields.size() > 1)
602 msg += 's';
603 msg += ' ';
604 msg += std::accumulate(newFields.begin(), newFields.end(), std::string{}, [](const auto &acc, const auto *field) {
605 return acc + (acc.length() ? ", " : "") + '`' + field->GetFieldName() + '`';
606 });
607 R__LOG_INFO(NTupleMergeLog()) << msg << ": adding " << (newFields.size() > 1 ? "them" : "it")
608 << " to the destination model (entry #" << mergeData.fNumDstEntries << ").";
609 }
610
611 changeset.fAddedFields.reserve(newFields.size());
612 // First add all non-projected fields...
613 for (const auto *fieldDesc : newFields) {
614 if (!fieldDesc->IsProjectedField()) {
615 auto field = fieldDesc->CreateField(*mergeData.fSrcDescriptor);
616 changeset.AddField(std::move(field));
617 }
618 }
619 // ...then add all projected fields.
620 for (const auto *fieldDesc : newFields) {
621 if (!fieldDesc->IsProjectedField())
622 continue;
623
625 auto field = fieldDesc->CreateField(*mergeData.fSrcDescriptor);
626 const auto sourceId = fieldDesc->GetProjectionSourceId();
627 const auto &sourceField = dstModel.GetConstField(mergeData.fSrcDescriptor->GetQualifiedFieldName(sourceId));
628 fieldMap[field.get()] = &sourceField;
629
630 for (const auto &subfield : *field) {
631 const auto &subFieldDesc = mergeData.fSrcDescriptor->GetFieldDescriptor(subfield.GetOnDiskId());
632 const auto subSourceId = subFieldDesc.GetProjectionSourceId();
633 const auto &subSourceField =
634 dstModel.GetConstField(mergeData.fSrcDescriptor->GetQualifiedFieldName(subSourceId));
636 }
637 changeset.fAddedProjectedFields.emplace_back(field.get());
639 }
640 dstModel.Freeze();
641 try {
642 mergeData.fDestination.UpdateSchema(changeset, mergeData.fNumDstEntries);
643 } catch (const ROOT::RException &ex) {
644 return R__FAIL(ex.GetError().GetReport());
645 }
646
647 commonFields.reserve(commonFields.size() + newFields.size());
648 for (const auto *field : newFields) {
649 const auto newFieldInDstId = mergeData.fDstDescriptor.FindFieldId(field->GetFieldName());
650 const auto &newFieldInDst = mergeData.fDstDescriptor.GetFieldDescriptor(newFieldInDstId);
651 commonFields.emplace_back(*field, newFieldInDst);
652 }
653
655}
656
657// Generates default (zero) values for the given columns
658[[nodiscard]]
660GenerateZeroPagesForColumns(size_t nEntriesToGenerate, std::span<const RColumnMergeInfo> columns,
663{
666
667 for (const auto &column : columns) {
668 const ROOT::RFieldDescriptor *field = column.fParentFieldDescriptor;
669
670 // Skip all auxiliary columns
671 assert(!field->GetLogicalColumnIds().empty());
672 if (field->GetLogicalColumnIds()[0] != column.fInputId)
673 continue;
674
675 // Check if this column is a child of a Collection or a Variant. If so, it has no data
676 // and can be skipped.
677 bool skipColumn = false;
678 auto nRepetitions = std::max<std::uint64_t>(field->GetNRepetitions(), 1);
679 for (auto parentId = field->GetParentId(); parentId != ROOT::kInvalidDescriptorId;) {
680 const ROOT::RFieldDescriptor &parent = column.fParentNTupleDescriptor->GetFieldDescriptor(parentId);
683 skipColumn = true;
684 break;
685 }
686 nRepetitions *= std::max<std::uint64_t>(parent.GetNRepetitions(), 1);
687 parentId = parent.GetParentId();
688 }
689 if (skipColumn)
690 continue;
691
692 const auto structure = field->GetStructure();
693
694 if (structure == ROOT::ENTupleStructure::kStreamer) {
695 return R__FAIL(
696 "Destination RNTuple contains a streamer field (" + field->GetFieldName() +
697 ") that is not present in one of the sources. "
698 "Creating a default value for a streamer field is ill-defined, therefore the merging process will abort.");
699 }
700
701 // NOTE: we cannot have a Record here because it has no associated columns.
703 structure == ROOT::ENTupleStructure::kPlain);
704
705 const auto &columnDesc = dstDescriptor.GetColumnDescriptor(column.fOutputId);
706 const auto colElement = RColumnElementBase::Generate(columnDesc.GetType());
708 const auto nBytesOnStorage = colElement->GetPackedSize(nElements);
709 // TODO(gparolini): make this configurable
710 constexpr auto kPageSizeLimit = 256 * 1024;
711 // TODO(gparolini): consider coalescing the last page if its size is less than some threshold
713 for (size_t i = 0; i < nPages; ++i) {
714 const auto pageSize = (i < nPages - 1) ? kPageSizeLimit : nBytesOnStorage - kPageSizeLimit * (nPages - 1);
716 const auto bufSize = pageSize + checksumSize;
717 assert(pageSize % colElement->GetSize() == 0);
718 const auto nElementsPerPage = pageSize / colElement->GetSize();
719 auto page = pageAlloc.NewPage(colElement->GetSize(), nElementsPerPage);
720 page.GrowUnchecked(nElementsPerPage);
721 memset(page.GetBuffer(), 0, page.GetNBytes());
722
723 auto &buffer = sealedPageData.fBuffers.emplace_back(new unsigned char[bufSize]);
725 sealConf.fElement = colElement.get();
726 sealConf.fPage = &page;
727 sealConf.fBuffer = buffer.get();
728 sealConf.fCompressionSettings = mergeData.fMergeOpts.fCompressionSettings.value();
729 sealConf.fWriteChecksum = mergeData.fDestination.GetWriteOptions().GetEnablePageChecksums();
731
732 sealedPageData.fPagesV.push_back({sealedPage});
733 sealedPageData.fGroups.emplace_back(column.fOutputId, sealedPageData.fPagesV.back().cbegin(),
734 sealedPageData.fPagesV.back().cend());
735 }
736 }
738}
739
740// Merges all columns appearing both in the source and destination RNTuples, just copying them if their
741// compression matches ("fast merge") or by unsealing and resealing them with the proper compression.
745 std::span<const RColumnMergeInfo> commonColumns,
749{
754
755 const RCluster *cluster = clusterPool.GetCluster(clusterDesc.GetId(), commonColumnSet);
756 // we expect the cluster pool to contain the requested set of columns, since they were
757 // validated by CompareDescriptorStructure().
759
760 for (size_t colIdx = 0; colIdx < nCommonColumnsInCluster; ++colIdx) {
761 const auto &column = commonColumns[colIdx];
762 const auto &columnId = column.fInputId;
763 R__ASSERT(clusterDesc.ContainsColumn(columnId));
764
765 const auto &columnDesc = mergeData.fSrcDescriptor->GetColumnDescriptor(columnId);
766 const auto srcColElement = column.fInMemoryType
767 ? ROOT::Internal::GenerateColumnElement(*column.fInMemoryType, columnDesc.GetType())
769 const auto dstColElement = column.fInMemoryType
770 ? ROOT::Internal::GenerateColumnElement(*column.fInMemoryType, column.fColumnType)
771 : RColumnElementBase::Generate(column.fColumnType);
772
773 // Now get the pages for this column in this cluster
774 const auto &pages = clusterDesc.GetPageRange(columnId);
775
777 sealedPages.resize(pages.GetPageInfos().size());
778
779 // Each column range potentially has a distinct compression settings
780 const auto colRangeCompressionSettings = clusterDesc.GetColumnRange(columnId).GetCompressionSettings().value();
781 // If either the compression or the encoding of the source doesn't match that of the destination, we need
782 // to reseal the page. Otherwise, if both match, we can fast merge.
783 const bool needsResealing =
784 colRangeCompressionSettings != mergeData.fMergeOpts.fCompressionSettings.value() ||
785 srcColElement->GetIdentifier().fOnDiskType != dstColElement->GetIdentifier().fOnDiskType;
786
787 if (needsResealing && mergeData.fMergeOpts.fExtraVerbose) {
789 << "Resealing column " << column.fColumnName << ": { compression: " << colRangeCompressionSettings << " => "
790 << mergeData.fMergeOpts.fCompressionSettings.value()
791 << ", onDiskType: " << RColumnElementBase::GetColumnTypeName(srcColElement->GetIdentifier().fOnDiskType)
792 << " => " << RColumnElementBase::GetColumnTypeName(dstColElement->GetIdentifier().fOnDiskType);
793 }
794
795 size_t pageBufferBaseIdx = sealedPageData.fBuffers.size();
796 // If the column range already has the right compression we don't need to allocate any new buffer, so we don't
797 // bother reserving memory for them.
798 if (needsResealing)
799 sealedPageData.fBuffers.resize(sealedPageData.fBuffers.size() + pages.GetPageInfos().size());
800
801 // If this column is deferred, we may need to fill "holes" until its real start. We fill any missing entry
802 // with zeroes, like we do for extraDstColumns.
803 // As an optimization, we don't do this for the first source (since we can rely on the FirstElementIndex and
804 // deferred column mechanism in that case).
805 // TODO: also avoid doing this if we added no real page of this column to the destination yet.
806 if (columnDesc.GetFirstElementIndex() > clusterDesc.GetFirstEntryIndex() && mergeData.fNumDstEntries > 0) {
807 const auto nMissingEntries = columnDesc.GetFirstElementIndex() - clusterDesc.GetFirstEntryIndex();
809 mergeData.fDstDescriptor, mergeData);
810 if (!res)
811 return R__FORWARD_ERROR(res);
812 }
813
814 // Loop over the pages
815 std::uint64_t pageIdx = 0;
816 for (const auto &pageInfo : pages.GetPageInfos()) {
817 assert(pageIdx < sealedPages.size());
818 assert(sealedPageData.fBuffers.size() == 0 || pageIdx < sealedPageData.fBuffers.size());
819 assert(pageInfo.GetLocator().GetType() != RNTupleLocator::kTypePageZero);
820
822 auto onDiskPage = cluster->GetOnDiskPage(key);
823
824 const auto checksumSize = pageInfo.HasChecksum() * RPageStorage::kNBytesPageChecksum;
826 sealedPage.SetNElements(pageInfo.GetNElements());
827 sealedPage.SetHasChecksum(pageInfo.HasChecksum());
828 sealedPage.SetBufferSize(pageInfo.GetLocator().GetNBytesOnStorage() + checksumSize);
829 sealedPage.SetBuffer(onDiskPage->GetAddress());
830 // TODO(gparolini): more graceful error handling (skip the page?)
831 sealedPage.VerifyChecksumIfEnabled().ThrowOnError();
832 R__ASSERT(onDiskPage && (onDiskPage->GetSize() == sealedPage.GetBufferSize()));
833
834 if (needsResealing) {
835 const auto uncompressedSize = srcColElement->GetSize() * sealedPage.GetNElements();
836 auto &buffer = sealedPageData.fBuffers[pageBufferBaseIdx + pageIdx];
838 RChangeCompressionFunc compressTask{
839 *srcColElement, *dstColElement, mergeData.fMergeOpts, sealedPage, *fPageAlloc, buffer.get(),
840 };
841
842 if (fTaskGroup)
843 fTaskGroup->Run(compressTask);
844 else
845 compressTask();
846 }
847
848 ++pageIdx;
849
850 } // end of loop over pages
851
852 if (fTaskGroup)
853 fTaskGroup->Wait();
854
855 sealedPageData.fPagesV.push_back(std::move(sealedPages));
856 sealedPageData.fGroups.emplace_back(column.fOutputId, sealedPageData.fPagesV.back().cbegin(),
857 sealedPageData.fPagesV.back().cend());
858 } // end loop over common columns
859
861}
862
863// Iterates over all clusters of `source` and merges their pages into `destination`.
864// It is assumed that all columns in `commonColumns` are present (and compatible) in both the source and
865// the destination's schemas.
866// The pages may be "fast-merged" (i.e. simply copied with no decompression/recompression) if the target
867// compression is unspecified or matches the original compression settings.
870 std::span<const RColumnMergeInfo> extraDstColumns, RNTupleMergeData &mergeData)
871{
873
874 std::vector<RColumnMergeInfo> missingColumns{extraDstColumns.begin(), extraDstColumns.end()};
875
876 // Loop over all clusters in this file.
877 // descriptor->GetClusterIterable() doesn't guarantee any specific order, so we explicitly
878 // request the first cluster.
879 ROOT::DescriptorId_t clusterId = mergeData.fSrcDescriptor->FindClusterId(0, 0);
881 const auto &clusterDesc = mergeData.fSrcDescriptor->GetClusterDescriptor(clusterId);
882 const auto nClusterEntries = clusterDesc.GetNEntries();
884
885 // NOTE: just because a column is in `commonColumns` it doesn't mean that each cluster in the source contains it,
886 // as it may be a deferred column that only has real data in a future cluster.
887 // We need to figure out which columns are actually present in this cluster so we only merge their pages (the
888 // missing columns are handled by synthesizing zero pages - see below).
890 while (nCommonColumnsInCluster > 0) {
891 // Since `commonColumns` is sorted by column input id, we can simply traverse it from the back and stop as
892 // soon as we find a common column that appears in this cluster: we know that in that case all previous
893 // columns must appear as well.
894 if (clusterDesc.ContainsColumn(commonColumns[nCommonColumnsInCluster - 1].fInputId))
895 break;
897 }
898
899 // Convert columns to a ColumnSet for the ClusterPool query
902 for (size_t i = 0; i < nCommonColumnsInCluster; ++i)
903 commonColumnSet.emplace(commonColumns[i].fInputId);
904
905 // For each cluster, the "missing columns" are the union of the extraDstColumns and the common columns
906 // that are not present in the cluster. We generate zero pages for all of them.
907 missingColumns.resize(extraDstColumns.size());
908 for (size_t i = nCommonColumnsInCluster; i < commonColumns.size(); ++i)
909 missingColumns.push_back(commonColumns[i]);
910
913 sealedPageData, mergeData, *fPageAlloc);
914 if (!res)
915 return R__FORWARD_ERROR(res);
916
918 mergeData.fDstDescriptor, mergeData);
919 if (!res)
920 return R__FORWARD_ERROR(res);
921
922 // Commit the pages and the clusters
923 mergeData.fDestination.CommitSealedPageV(sealedPageData.fGroups);
924 mergeData.fDestination.CommitCluster(nClusterEntries);
925 mergeData.fNumDstEntries += nClusterEntries;
926
927 // Go to the next cluster
928 clusterId = mergeData.fSrcDescriptor->FindNextClusterId(clusterId);
929 }
930
931 // TODO(gparolini): when we get serious about huge file support (>~ 100GB) we might want to check here
932 // the size of the running page list and commit a cluster group when it exceeds some threshold,
933 // which would prevent the page list from getting too large.
934 // However, as of today, we aren't really handling such huge files, and even relatively big ones
935 // such as the CMS dataset have a page list size of about only 2 MB.
936 // So currently we simply merge all cluster groups into one.
938}
939
940static std::optional<std::type_index> ColumnInMemoryType(std::string_view fieldType, ENTupleColumnType onDiskType)
941{
944 return typeid(ROOT::Internal::RColumnIndex);
945
947 return typeid(ROOT::Internal::RColumnSwitch);
948
949 // clang-format off
950 if (fieldType == "bool") return typeid(bool);
951 if (fieldType == "std::byte") return typeid(std::byte);
952 if (fieldType == "char") return typeid(char);
953 if (fieldType == "std::int8_t") return typeid(std::int8_t);
954 if (fieldType == "std::uint8_t") return typeid(std::uint8_t);
955 if (fieldType == "std::int16_t") return typeid(std::int16_t);
956 if (fieldType == "std::uint16_t") return typeid(std::uint16_t);
957 if (fieldType == "std::int32_t") return typeid(std::int32_t);
958 if (fieldType == "std::uint32_t") return typeid(std::uint32_t);
959 if (fieldType == "std::int64_t") return typeid(std::int64_t);
960 if (fieldType == "std::uint64_t") return typeid(std::uint64_t);
961 if (fieldType == "float") return typeid(float);
962 if (fieldType == "double") return typeid(double);
963 // clang-format on
964
965 // if the type is not one of those above, we use the default in-memory type.
966 return std::nullopt;
967}
968
969// Given a field, fill `columns` and `mergeData.fColumnIdMap` with information about all columns belonging to it and its
970// subfields. `mergeData.fColumnIdMap` is used to map matching columns from different sources to the same output column
971// in the destination. We match columns by their "fully qualified name", which is the concatenation of their ancestor
972// fields' names and the column index. By this point, since we called `CompareDescriptorStructure()` earlier, we should
973// be guaranteed that two matching columns will have at least compatible representations. NOTE: srcFieldDesc and
974// dstFieldDesc may alias.
975static void AddColumnsFromField(std::vector<RColumnMergeInfo> &columns, const ROOT::RNTupleDescriptor &srcDesc,
977 const ROOT::RFieldDescriptor &dstFieldDesc, const std::string &prefix = "")
978{
979 std::string name = prefix + '.' + srcFieldDesc.GetFieldName();
980
981 const auto &columnIds = srcFieldDesc.GetLogicalColumnIds();
982 columns.reserve(columns.size() + columnIds.size());
983 // NOTE: here we can match the src and dst columns by column index because we forbid merging fields with
984 // different column representations.
985 for (auto i = 0u; i < srcFieldDesc.GetLogicalColumnIds().size(); ++i) {
986 // We don't want to try and merge alias columns
987 if (srcFieldDesc.IsProjectedField())
988 continue;
989
990 auto srcColumnId = srcFieldDesc.GetLogicalColumnIds()[i];
991 const auto &srcColumn = srcDesc.GetColumnDescriptor(srcColumnId);
992
994 info.fColumnName = name + '.' + std::to_string(srcColumn.GetIndex());
995 info.fInputId = srcColumn.GetPhysicalId();
996 // NOTE(gparolini): the parent field is used when synthesizing zero pages, which happens in 2 situations:
997 // 1. when adding extra dst columns (in which case we need to synthesize zero pages for the incoming src), and
998 // 2. when merging a deferred column into an existing column (in which case we need to fill the "hole" with
999 // zeroes). For the first case srcFieldDesc and dstFieldDesc are the same (see the calling site of this function),
1000 // but for the second case they're not, and we need to pick the source field because we will then check the
1001 // column's *input* id inside fParentFieldDescriptor to see if it's a suppressed column (see
1002 // GenerateZeroPagesForColumns()).
1003 info.fParentFieldDescriptor = &srcFieldDesc;
1004 // Save the parent field descriptor since this may be either the source or destination descriptor depending on
1005 // whether this is an extraDstField or a commonField. We will need this in GenerateZeroPagesForColumns() to
1006 // properly walk up the field hierarchy.
1007 info.fParentNTupleDescriptor = &srcDesc;
1008
1009 if (auto it = mergeData.fColumnIdMap.find(info.fColumnName); it != mergeData.fColumnIdMap.end()) {
1010 info.fOutputId = it->second.fColumnId;
1011 info.fColumnType = it->second.fColumnType;
1012 } else {
1013 info.fOutputId = mergeData.fColumnIdMap.size();
1014 // NOTE(gparolini): map the type of src column to the type of dst column.
1015 // This mapping is only relevant for common columns and it's done to ensure we keep a consistent
1016 // on-disk representation of the same column.
1017 // This is also important to do for first source when it is used to generate the destination sink,
1018 // because even in that case their column representations may differ.
1019 // e.g. if the destination has a different compression than the source, an integer column might be
1020 // zigzag-encoded in the source but not in the destination.
1021 auto dstColumnId = dstFieldDesc.GetLogicalColumnIds()[i];
1022 const auto &dstColumn = mergeData.fDstDescriptor.GetColumnDescriptor(dstColumnId);
1023 info.fColumnType = dstColumn.GetType();
1024 mergeData.fColumnIdMap[info.fColumnName] = {info.fOutputId, info.fColumnType};
1025 }
1026
1027 if (mergeData.fMergeOpts.fExtraVerbose) {
1028 R__LOG_INFO(NTupleMergeLog()) << "Adding column " << info.fColumnName << " with log.id " << srcColumnId
1029 << ", phys.id " << srcColumn.GetPhysicalId() << ", type "
1030 << RColumnElementBase::GetColumnTypeName(srcColumn.GetType()) << " -> log.id "
1031 << info.fOutputId << ", type "
1033 }
1034
1035 // Since we disallow merging fields of different types, src and dstFieldDesc must have the same type name.
1036 assert(srcFieldDesc.GetTypeName() == dstFieldDesc.GetTypeName());
1037 info.fInMemoryType = ColumnInMemoryType(srcFieldDesc.GetTypeName(), info.fColumnType);
1038 columns.emplace_back(info);
1039 }
1040
1041 const auto &srcChildrenIds = srcFieldDesc.GetLinkIds();
1042 const auto &dstChildrenIds = dstFieldDesc.GetLinkIds();
1043 assert(srcChildrenIds.size() == dstChildrenIds.size());
1044 for (auto i = 0u; i < srcChildrenIds.size(); ++i) {
1045 const auto &srcChild = srcDesc.GetFieldDescriptor(srcChildrenIds[i]);
1046 const auto &dstChild = mergeData.fDstDescriptor.GetFieldDescriptor(dstChildrenIds[i]);
1048 }
1049}
1050
1051// Converts the fields comparison data to the corresponding column information.
1052// While doing so, it collects such information in `mergeData.fColumnIdMap`, which is used by later calls to this
1053// function to map already-seen column names to their chosen outputId, type and so on.
1054static RColumnInfoGroup GatherColumnInfos(const RDescriptorsComparison &descCmp, const ROOT::RNTupleDescriptor &srcDesc,
1056{
1057 RColumnInfoGroup res;
1058 for (const ROOT::RFieldDescriptor *field : descCmp.fExtraDstFields) {
1059 AddColumnsFromField(res.fExtraDstColumns, mergeData.fDstDescriptor, mergeData, *field, *field);
1060 }
1061 for (const auto &[srcField, dstField] : descCmp.fCommonFields) {
1062 AddColumnsFromField(res.fCommonColumns, srcDesc, mergeData, *srcField, *dstField);
1063 }
1064
1065 // Sort the commonColumns by ID so we can more easily tell how many common columns each cluster has
1066 // (since each cluster must contain all columns of the previous cluster plus potentially some new ones)
1067 std::sort(res.fCommonColumns.begin(), res.fCommonColumns.end(),
1068 [](const auto &a, const auto &b) { return a.fInputId < b.fInputId; });
1069
1070 return res;
1071}
1072
1074 ColumnIdMap_t &colIdMap, const std::string &prefix = "")
1075{
1076 std::string name = prefix + '.' + fieldDesc.GetFieldName();
1077 for (const auto &colId : fieldDesc.GetLogicalColumnIds()) {
1078 const auto &colDesc = desc.GetColumnDescriptor(colId);
1079 RColumnOutInfo info{};
1080 const auto colName = name + '.' + std::to_string(colDesc.GetIndex());
1081 info.fColumnId = colDesc.GetLogicalId();
1082 info.fColumnType = colDesc.GetType();
1084 }
1085
1086 for (const auto &subId : fieldDesc.GetLinkIds()) {
1087 const auto &subfield = desc.GetFieldDescriptor(subId);
1089 }
1090}
1091
1092RNTupleMerger::RNTupleMerger(std::unique_ptr<ROOT::Internal::RPagePersistentSink> destination,
1093 std::unique_ptr<ROOT::RNTupleModel> model)
1094 // TODO(gparolini): consider using an arena allocator instead, since we know the precise lifetime
1095 // of the RNTuples we are going to handle (e.g. we can reset the arena at every source)
1096 : fDestination(std::move(destination)),
1097 fPageAlloc(std::make_unique<ROOT::Internal::RPageAllocatorHeap>()),
1098 fModel(std::move(model))
1099{
1101
1102#ifdef R__USE_IMT
1105#endif
1106}
1107
1108RNTupleMerger::RNTupleMerger(std::unique_ptr<ROOT::Internal::RPagePersistentSink> destination)
1109 : RNTupleMerger(std::move(destination), nullptr)
1110{
1111}
1112
1114{
1116
1118
1119 // Set compression settings if unset and verify it's compatible with the sink
1120 {
1121 const auto dstCompSettings = fDestination->GetWriteOptions().GetCompression();
1122 if (!mergeOpts.fCompressionSettings) {
1123 mergeOpts.fCompressionSettings = dstCompSettings;
1124 } else if (*mergeOpts.fCompressionSettings != dstCompSettings) {
1125 return R__FAIL(std::string("The compression given to RNTupleMergeOptions is different from that of the "
1126 "sink! (opts: ") +
1127 std::to_string(*mergeOpts.fCompressionSettings) + ", sink: " + std::to_string(dstCompSettings) +
1128 ") This is currently unsupported.");
1129 }
1130 }
1131
1132 // we should have a model if and only if the destination is initialized.
1133 if (!!fModel != fDestination->IsInitialized()) {
1134 return R__FAIL(
1135 "passing an already-initialized destination to RNTupleMerger::Merge (i.e. trying to do incremental "
1136 "merging) can only be done by providing a valid ROOT::RNTupleModel when constructing the RNTupleMerger.");
1137 }
1138
1140 mergeData.fNumDstEntries = mergeData.fDestination.GetNEntries();
1141
1142 if (fModel) {
1143 // If this is an incremental merging, pre-fill the column id map with the existing destination ids.
1144 // Otherwise we would generate new output ids that may not match the ones in the destination!
1145 for (const auto &field : mergeData.fDstDescriptor.GetTopLevelFields()) {
1146 PrefillColumnMap(fDestination->GetDescriptor(), field, mergeData.fColumnIdMap);
1147 }
1148 }
1149
1150#define SKIP_OR_ABORT(errMsg) \
1151 do { \
1152 if (mergeOpts.fErrBehavior == ENTupleMergeErrBehavior::kSkip) { \
1153 R__LOG_WARNING(NTupleMergeLog()) << "Skipping RNTuple due to: " << (errMsg); \
1154 continue; \
1155 } else { \
1156 return R__FAIL(errMsg); \
1157 } \
1158 } while (0)
1159
1160 // Merge main loop
1161 for (RPageSource *source : sources) {
1162 source->Attach(RNTupleSerializer::EDescriptorDeserializeMode::kForWriting);
1163 auto srcDescriptor = source->GetSharedDescriptorGuard();
1164 mergeData.fSrcDescriptor = &srcDescriptor.GetRef();
1165
1166 // Create sink from the input model if not initialized
1167 if (!fModel) {
1168 fModel = fDestination->InitFromDescriptor(srcDescriptor.GetRef(), false /* copyClusters */);
1169 }
1170
1171 for (const auto &extraTypeInfoDesc : srcDescriptor->GetExtraTypeInfoIterable())
1172 fDestination->UpdateExtraTypeInfo(extraTypeInfoDesc);
1173
1174 auto descCmpRes = CompareDescriptorStructure(mergeData.fDstDescriptor, srcDescriptor.GetRef());
1175 if (!descCmpRes) {
1176 SKIP_OR_ABORT(std::string("Source RNTuple has an incompatible schema with the destination:\n") +
1177 descCmpRes.GetError()->GetReport());
1178 }
1179 auto descCmp = descCmpRes.Unwrap();
1180
1181 // If the current source is missing some fields and we're not in Union mode, error
1182 // (if we are in Union mode, MergeSourceClusters will fill the missing fields with default values).
1183 if (mergeOpts.fMergingMode != ENTupleMergingMode::kUnion && !descCmp.fExtraDstFields.empty()) {
1184 std::string msg = "Source RNTuple is missing the following fields:";
1185 for (const auto *field : descCmp.fExtraDstFields) {
1186 msg += "\n " + field->GetFieldName() + " : " + field->GetTypeName();
1187 }
1189 }
1190
1191 // handle extra src fields
1192 if (descCmp.fExtraSrcFields.size()) {
1193 if (mergeOpts.fMergingMode == ENTupleMergingMode::kUnion) {
1194 // late model extension for all fExtraSrcFields in Union mode
1195 auto res = ExtendDestinationModel(descCmp.fExtraSrcFields, *fModel, mergeData, descCmp.fCommonFields);
1196 if (!res)
1197 return R__FORWARD_ERROR(res);
1198 } else if (mergeOpts.fMergingMode == ENTupleMergingMode::kStrict) {
1199 // If the current source has extra fields and we're in Strict mode, error
1200 std::string msg = "Source RNTuple has extra fields that the destination RNTuple doesn't have:";
1201 for (const auto *field : descCmp.fExtraSrcFields) {
1202 msg += "\n " + field->GetFieldName() + " : " + field->GetTypeName();
1203 }
1205 }
1206 }
1207
1208 // handle extra dst fields & common fields
1210 auto res = MergeSourceClusters(*source, columnInfos.fCommonColumns, columnInfos.fExtraDstColumns, mergeData);
1211 if (!res)
1212 return R__FORWARD_ERROR(res);
1213 } // end loop over sources
1214
1215 if (fDestination->GetNEntries() == 0)
1216 R__LOG_WARNING(NTupleMergeLog()) << "Output RNTuple '" << fDestination->GetNTupleName() << "' has no entries.";
1217
1218 // Commit the output
1219 fDestination->CommitClusterGroup();
1220 fDestination->CommitDataset();
1221
1222 return RResult<void>::Success();
1223}
fBuffer
#define R__FORWARD_ERROR(res)
Short-hand to return an RResult<T> in an error state (i.e. after checking)
Definition RError.hxx:304
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
Definition RError.hxx:300
#define R__LOG_WARNING(...)
Definition RLogger.hxx:358
#define R__LOG_ERROR(...)
Definition RLogger.hxx:357
#define R__LOG_INFO(...)
Definition RLogger.hxx:359
static std::optional< std::type_index > ColumnInMemoryType(std::string_view fieldType, ENTupleColumnType onDiskType)
static ROOT::RResult< RDescriptorsComparison > CompareDescriptorStructure(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTupleDescriptor &src)
Compares the top level fields of dst and src and determines whether they can be merged or not.
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)
static std::optional< ENTupleMergeErrBehavior > ParseOptionErrBehavior(const TString &opts)
static ROOT::RLogChannel & NTupleMergeLog()
#define SKIP_OR_ABORT(errMsg)
static std::optional< T > ParseStringOption(const TString &opts, const char *pattern, std::initializer_list< std::pair< const char *, T > > validValues)
static bool IsSplitOrUnsplitVersionOf(ENTupleColumnType a, ENTupleColumnType b)
static void AddColumnsFromField(std::vector< RColumnMergeInfo > &columns, const ROOT::RNTupleDescriptor &srcDesc, RNTupleMergeData &mergeData, const ROOT::RFieldDescriptor &srcFieldDesc, const ROOT::RFieldDescriptor &dstFieldDesc, const std::string &prefix="")
static std::optional< ENTupleMergingMode > ParseOptionMergingMode(const TString &opts)
static void PrefillColumnMap(const ROOT::RNTupleDescriptor &desc, const ROOT::RFieldDescriptor &fieldDesc, ColumnIdMap_t &colIdMap, const std::string &prefix="")
static RColumnInfoGroup GatherColumnInfos(const RDescriptorsComparison &descCmp, const ROOT::RNTupleDescriptor &srcDesc, RNTupleMergeData &mergeData)
static ROOT::RResult< void > ExtendDestinationModel(std::span< const ROOT::RFieldDescriptor * > newFields, ROOT::RNTupleModel &dstModel, RNTupleMergeData &mergeData, std::vector< RCommonField > &commonFields)
static bool BeginsWithDelimitedWord(const TString &str, const char *word)
#define b(i)
Definition RSha256.hxx:100
#define a(i)
Definition RSha256.hxx:99
size_t size(const MatrixT &matrix)
retrieve the size of a square matrix
ROOT::Detail::TRangeCast< T, true > TRangeDynCast
TRangeDynCast is an adapter class that allows the typed iteration through a TCollection.
#define R__ASSERT(e)
Checks condition e and reports a fatal error if it's false.
Definition TError.h:125
Option_t Option_t TPoint TPoint const char GetTextMagnitude GetFillStyle GetLineColor GetLineWidth GetMarkerStyle GetTextAlign GetTextColor GetTextSize void value
Option_t Option_t TPoint TPoint const char GetTextMagnitude GetFillStyle GetLineColor GetLineWidth GetMarkerStyle GetTextAlign GetTextColor GetTextSize void char Point_t Rectangle_t src
char name[80]
Definition TGX11.cxx:110
TRObject operator()(const T1 &t1) const
The available trivial, native content types of a column.
Given a set of RPageSources merge them into an RPagePersistentSink, optionally changing their compres...
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)
std::unique_ptr< ROOT::RNTupleModel > fModel
RNTupleMerger(std::unique_ptr< ROOT::Internal::RPagePersistentSink > destination, std::unique_ptr< ROOT::RNTupleModel > model)
Creates a RNTupleMerger with the given destination.
ROOT::RResult< void > MergeSourceClusters(ROOT::Internal::RPageSource &source, std::span< const RColumnMergeInfo > commonColumns, std::span< const RColumnMergeInfo > extraDstColumns, RNTupleMergeData &mergeData)
std::unique_ptr< ROOT::Internal::RPagePersistentSink > fDestination
RResult< void > Merge(std::span< ROOT::Internal::RPageSource * > sources, const RNTupleMergeOptions &mergeOpts=RNTupleMergeOptions())
Merge a given set of sources into the destination.
A class to manage the asynchronous execution of work items.
Managed a set of clusters containing compressed and packed pages.
An in-memory subset of the packed and compressed pages of a cluster.
Definition RCluster.hxx:148
std::unordered_set< ROOT::DescriptorId_t > ColumnSet_t
Definition RCluster.hxx:150
A column element encapsulates the translation between basic C++ types and their column representation...
static const char * GetColumnTypeName(ROOT::ENTupleColumnType type)
static std::unique_ptr< RColumnElementBase > Generate(ROOT::ENTupleColumnType type)
If CppT == void, use the default C++ type for the given column type.
The in-memory representation of a 32bit or 64bit on-disk index column.
Holds the index and the tag of a kSwitch column.
A helper class for serializing and deserialization of the RNTuple binary format.
Uses standard C++ memory allocation for the column data pages.
Abstract interface to allocate and release pages.
Abstract interface to write data into an ntuple.
RSealedPage SealPage(const ROOT::Internal::RPage &page, const ROOT::Internal::RColumnElementBase &element)
Helper for streaming a page.
Storage provider that reads ntuple pages from a file.
static std::unique_ptr< RPageSourceFile > CreateFromAnchor(const RNTuple &anchor, const ROOT::RNTupleReadOptions &options=ROOT::RNTupleReadOptions())
Used from the RNTuple class to build a datasource if the anchor is already available.
Abstract interface to read data from an ntuple.
Common functionality of an ntuple storage for both reading and writing.
static constexpr std::size_t kNBytesPageChecksum
The page checksum is a 64bit xxhash3.
std::deque< RSealedPage > SealedPageSequence_t
std::unordered_map< const ROOT::RFieldBase *, const ROOT::RFieldBase * > FieldMap_t
The map keys are the projected target fields, the map values are the backing source fields Note that ...
RResult< void > Add(std::unique_ptr< ROOT::RFieldBase > field, const FieldMap_t &fieldMap)
Adds a new projected field.
Metadata for RNTuple clusters.
Base class for all ROOT issued exceptions.
Definition RError.hxx:79
Metadata stored for every field of an RNTuple.
ROOT::ENTupleStructure GetStructure() const
ROOT::DescriptorId_t GetParentId() const
std::uint64_t GetNRepetitions() const
A log configuration for a channel, e.g.
Definition RLogger.hxx:98
The on-storage metadata of an RNTuple.
const RColumnDescriptor & GetColumnDescriptor(ROOT::DescriptorId_t columnId) const
const RFieldDescriptor & GetFieldDescriptor(ROOT::DescriptorId_t fieldId) const
The RNTupleModel encapulates the schema of an RNTuple.
Common user-tunable settings for storing RNTuples.
Representation of an RNTuple data set in a ROOT file.
Definition RNTuple.hxx:67
Long64_t Merge(TCollection *input, TFileMergeInfo *mergeInfo)
RNTuple implements the hadd MergeFile interface Merge this NTuple with the input list entries.
const_iterator begin() const
const_iterator end() const
The class is used as a return type for operations that can fail; wraps a value of type T or an RError...
Definition RError.hxx:198
Collection abstract base class.
Definition TCollection.h:65
A ROOT file is an on-disk file, usually with extension .root, that stores objects in a file-system-li...
Definition TFile.h:131
Book space in a file, create I/O buffers, to fill them, (un)compress them.
Definition TKey.h:28
Mother of all ROOT objects.
Definition TObject.h:41
Basic string class.
Definition TString.h:138
@ kIgnoreCase
Definition TString.h:285
Double_t x[n]
Definition legend1.C:17
Double_t ex[n]
Definition legend1.C:17
@ kStrict
The merger will refuse to merge any 2 RNTuples whose schema doesn't match exactly.
@ kUnion
The merger will update the output model to include all columns from all sources.
std::ostream & operator<<(std::ostream &os, const std::optional< ROOT::RColumnDescriptor::RValueRange > &x)
std::unique_ptr< T[]> MakeUninitArray(std::size_t size)
Make an array of default-initialized elements.
RProjectedFields & GetProjectedFieldsOfModel(RNTupleModel &model)
std::unique_ptr< RColumnElementBase > GenerateColumnElement(std::type_index inMemoryType, ROOT::ENTupleColumnType onDiskType)
Namespace for new ROOT classes and functions.
Bool_t IsImplicitMTEnabled()
Returns true if the implicit multi-threading in ROOT is enabled.
Definition TROOT.cxx:600
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
std::uint64_t NTupleSize_t
Integer type long enough to hold the maximum number of entries in a column.
constexpr DescriptorId_t kInvalidDescriptorId
ENTupleColumnType
const ROOT::RFieldDescriptor * fParentFieldDescriptor
std::optional< std::type_index > fInMemoryType
const ROOT::RNTupleDescriptor * fParentNTupleDescriptor
const ROOT::RNTupleDescriptor * fSrcDescriptor
RNTupleMergeData(std::span< RPageSource * > sources, RPageSink &destination, const RNTupleMergeOptions &mergeOpts)
const ROOT::RNTupleDescriptor & fDstDescriptor
Set of merging options to pass to RNTupleMerger.
std::optional< std::uint32_t > fCompressionSettings
If fCompressionSettings is empty (the default), the merger will not change the compression of any of ...
std::vector< RPageStorage::RSealedPageGroup > fGroups
std::deque< RPageStorage::SealedPageSequence_t > fPagesV
std::vector< std::unique_ptr< std::uint8_t[]> > fBuffers
The incremental changes to a RNTupleModel
On-disk pages within a page source are identified by the column and page number.
Definition RCluster.hxx:51
Parameters for the SealPage() method.
A sealed page contains the bytes of a page as written to storage (packed & compressed).
@ kUseGeneralPurpose
Use the new recommended general-purpose setting; it is a best trade-off between compression ratio/dec...
Definition Compression.h:58