40 if (!inputs || inputs->
GetEntries() < 3 || !mergeInfo)
47 std::string ntupleName = std::string(itr()->GetName());
55 TKey *outKey = outFile->
FindKey(ntupleName.c_str());
60 Error(
"RNTuple::Merge",
"Output file already has key, but not of type RNTuple!");
68 const int compression =
75 auto destination = std::make_unique<Internal::RPageSinkFile>(ntupleName, *outFile, writeOpts);
81 auto desc = source->GetSharedDescriptorGuard();
82 destination->InitFromDescriptor(desc.GetRef());
86 std::vector<std::unique_ptr<Internal::RPageSourceFile>> sources;
87 std::vector<Internal::RPageSource *> sourcePtrs;
89 while (
const auto &pitr = itr()) {
98 sourcePtrs.reserve(sources.size());
99 for (
const auto &s : sources) {
100 sourcePtrs.push_back(s.get());
107 merger.
Merge(sourcePtrs, *destination, options);
111 *
this = *outFile->
Get<
RNTuple>(ntupleName.c_str());
115 Error(
"RNTuple::Merge",
"Exception thrown while merging: %s",
ex.what());
121 std::vector<ROOT::Experimental::Internal::RNTupleMerger::RColumnInfo> &columns)
123 for (
auto &column : columns) {
124 column.fColumnOutputId = fOutputIdMap.size();
125 fOutputIdMap[column.fColumnName +
"." + column.fColumnTypeAndVersion] = column.fColumnOutputId;
131 std::vector<ROOT::Experimental::Internal::RNTupleMerger::RColumnInfo> &columns)
134 if (fOutputIdMap.size() != columns.size()) {
138 for (
auto &column : columns) {
140 column.fColumnOutputId = fOutputIdMap.at(column.fColumnName +
"." + column.fColumnTypeAndVersion);
141 }
catch (
const std::out_of_range &) {
142 throw RException(
R__FAIL(
"Column NOT found in the first source w/ name " + column.fColumnName +
143 " type and version " + column.fColumnTypeAndVersion));
150 std::vector<RColumnInfo> &columns)
153 AddColumnsFromField(columns, descriptor, descriptor.
GetFieldZero());
156 if (fOutputIdMap.empty()) {
157 BuildColumnIdMap(columns);
159 ValidateColumns(columns);
167 const std::string &prefix)
170 std::string
name = prefix + field.GetFieldName() +
".";
171 const std::string typeAndVersion = field.GetTypeName() +
"." + std::to_string(field.GetTypeVersion());
173 columns.reserve(columns.size() + columnIter.count());
174 for (
const auto &column : columnIter) {
175 columns.emplace_back(
name + std::to_string(column.GetIndex()), typeAndVersion, column.GetPhysicalId(),
178 AddColumnsFromField(columns, desc, field,
name);
186 std::vector<RColumnInfo> columns;
193 std::unique_ptr<RNTupleModel> model;
194 std::optional<TTaskGroup> taskGroup;
201 for (
const auto &source : sources) {
207 auto descriptor = source->GetSharedDescriptorGuard();
211 columns.clear(), columnSet.clear();
212 CollectColumns(descriptor.GetRef(), columns);
213 columnSet.reserve(columns.size());
214 for (
const auto &column : columns)
215 columnSet.emplace(column.fColumnInputId);
219 model = descriptor->CreateModel();
220 destination.
Init(*model.get());
223 for (
const auto &extraTypeInfoDesc : descriptor->GetExtraTypeInfoIterable()) {
228 if (source->GetNEntries() == 0) {
235 auto clusterId = descriptor->FindClusterId(0, 0);
238 auto *cluster = clusterPool.GetCluster(clusterId, columnSet);
240 const auto &clusterDesc = descriptor->GetClusterDescriptor(clusterId);
244 std::deque<RPageStorage::SealedPageSequence_t> sealedPagesV;
245 std::vector<RPageStorage::RSealedPageGroup> sealedPageGroups;
246 std::vector<std::unique_ptr<unsigned char[]>> sealedPageBuffers;
248 for (
const auto &column : columns) {
252 auto columnId = column.fColumnInputId;
253 if (!clusterDesc.ContainsColumn(columnId)) {
257 const auto &columnDesc = descriptor->GetColumnDescriptor(columnId);
258 const auto colElement = RColumnElementBase::Generate(columnDesc.GetType());
261 const auto &pages = clusterDesc.GetPageRange(columnId);
264 sealedPages.resize(pages.fPageInfos.size());
267 const auto colRangeCompressionSettings = clusterDesc.GetColumnRange(columnId).fCompressionSettings;
273 size_t pageBufferBaseIdx = sealedPageBuffers.size();
274 if (colRangeCompressionSettings != 0)
275 sealedPageBuffers.resize(sealedPageBuffers.size() + pages.fPageInfos.size());
277 std::uint64_t pageIdx = 0;
280 for (
const auto &pageInfo : pages.fPageInfos) {
281 assert(pageIdx < sealedPages.size());
282 assert(sealedPageBuffers.size() == 0 || pageIdx < sealedPageBuffers.size());
285 auto onDiskPage = cluster->GetOnDiskPage(key);
287 const auto checksumSize = pageInfo.fHasChecksum * RPageStorage::kNBytesPageChecksum;
291 sealedPage.
SetBufferSize(pageInfo.fLocator.fBytesOnStorage + checksumSize);
292 sealedPage.
SetBuffer(onDiskPage->GetAddress());
297 if (needsCompressionChange) {
299 pageIdx, colRangeCompressionSettings, pageBufferBaseIdx, checksumSize,
301 &colElement, &pageInfo, &options,
303 &sealedPage, &sealedPageBuffers]() {
309 const auto uncompressedSize = colElement->GetSize() * sealedPage.GetNElements();
310 auto zipBuffer = std::make_unique<unsigned char[]>(uncompressedSize);
311 RNTupleDecompressor::Unzip(sealedPage.GetBuffer(), sealedPage.GetDataSize(), uncompressedSize,
315 if (uncompressedSize != sealedPage.GetDataSize()) {
317 R__ASSERT(colRangeCompressionSettings != 0);
323 R__ASSERT(sealedPage.GetDataSize() < uncompressedSize);
324 auto &newBuf = sealedPageBuffers[pageBufferBaseIdx + pageIdx];
325 newBuf = std::make_unique<unsigned char[]>(uncompressedSize + checksumSize);
326 sealedPage.SetBuffer(newBuf.get());
335 const auto newNBytes =
336 RNTupleCompressor::Zip(zipBuffer.get(), uncompressedSize, options.fCompressionSettings,
337 const_cast<void *
>(sealedPage.GetBuffer()));
338 sealedPage.SetBufferSize(newNBytes + checksumSize);
339 if (pageInfo.fHasChecksum) {
341 sealedPage.ChecksumIfEnabled();
346 taskGroup->Run(taskFunc);
357 sealedPagesV.push_back(std::move(sealedPages));
358 sealedPageGroups.emplace_back(column.fColumnOutputId, sealedPagesV.back().cbegin(),
359 sealedPagesV.back().cend());
370 clusterId = descriptor->FindNextClusterId(clusterId);
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
#define R__ASSERT(e)
Checks condition e and reports a fatal error if it's false.
void Error(const char *location, const char *msgfmt,...)
Use this function in case an error occurred.
Managed a set of clusters containing compressed and packed pages.
std::unordered_set< DescriptorId_t > ColumnSet_t
Given a set of RPageSources merge them into an RPageSink, optionally changing their compression.
void AddColumnsFromField(std::vector< RColumnInfo > &columns, const RNTupleDescriptor &desc, const RFieldDescriptor &fieldDesc, const std::string &prefix="")
Recursively add columns from a given field.
void Merge(std::span< RPageSource * > sources, RPageSink &destination, const RNTupleMergeOptions &options=RNTupleMergeOptions())
Merge a given set of sources into the destination.
void CollectColumns(const RNTupleDescriptor &descriptor, std::vector< RColumnInfo > &columns)
Recursively collect all the columns for all the fields rooted at field zero.
void BuildColumnIdMap(std::vector< RColumnInfo > &columns)
Build the internal column id map from the first source This is where we assign the output ids for the...
void ValidateColumns(std::vector< RColumnInfo > &columns)
Validate the columns against the internal map that is built from the first source This is where we as...
Abstract interface to write data into an ntuple.
bool IsInitialized() const
virtual void UpdateExtraTypeInfo(const RExtraTypeInfoDescriptor &extraTypeInfo)=0
Adds an extra type information record to schema.
void CommitDataset()
Run the registered callbacks and finalize the current cluster and the entrire data set.
virtual const RNTupleDescriptor & GetDescriptor() const =0
Return the RNTupleDescriptor being constructed.
void Init(RNTupleModel &model)
Physically creates the storage container to hold the ntuple (e.g., a keys a TFile or an S3 bucket) In...
virtual void CommitClusterGroup()=0
Write out the page locations (page list envelope) for all the committed clusters since the last call ...
virtual std::uint64_t CommitCluster(NTupleSize_t nNewEntries)=0
Finalize the current cluster and create a new one for the following data.
virtual void CommitSealedPageV(std::span< RPageStorage::RSealedPageGroup > ranges)=0
Write a vector of preprocessed pages to storage. The corresponding columns must have been added befor...
static std::unique_ptr< RPageSourceFile > CreateFromAnchor(const RNTuple &anchor, const RNTupleReadOptions &options=RNTupleReadOptions())
Used from the RNTuple class to build a datasource if the anchor is already available.
std::deque< RSealedPage > SealedPageSequence_t
Base class for all ROOT issued exceptions.
Meta-data stored for every field of an ntuple.
The on-storage meta-data of an ntuple.
RColumnDescriptorIterable GetColumnIterable() const
const RFieldDescriptor & GetFieldZero() const
RFieldDescriptorIterable GetFieldIterable(const RFieldDescriptor &fieldDesc) const
Common user-tunable settings for storing ntuples.
void SetCompression(int val)
void SetUseBufferedWrite(bool val)
Representation of an RNTuple data set in a ROOT file.
Long64_t Merge(TCollection *input, TFileMergeInfo *mergeInfo)
RNTuple implements the hadd MergeFile interface Merge this NTuple with the input list entries.
void ThrowOnError()
Short-hand method to throw an exception in the case of errors.
A class to manage the asynchronous execution of work items.
Collection abstract base class.
virtual Int_t GetEntries() const
TKey * FindKey(const char *keyname) const override
Find key with name keyname in the current directory.
TObject * Get(const char *namecycle) override
Return pointer to object identified by namecycle.
A ROOT file is an on-disk file, usually with extension .root, that stores objects in a file-system-li...
Int_t GetCompressionSettings() const
Book space in a file, create I/O buffers, to fill them, (un)compress them.
T * ReadObject()
To read an object (non deriving from TObject) from the file.
Bool_t Contains(const char *pat, ECaseCompare cmp=kExact) const
constexpr int kUnknownCompressionSettings
constexpr DescriptorId_t kInvalidDescriptorId
Bool_t IsImplicitMTEnabled()
Returns true if the implicit multi-threading in ROOT is enabled.
int fCompressionSettings
If fCompressionSettings == kUnknownCompressionSettings (the default), the merger will not change the ...
On-disk pages within a page source are identified by the column and page number.
A sealed page contains the bytes of a page as written to storage (packed & compressed).
void SetHasChecksum(bool hasChecksum)
void SetNElements(std::uint32_t nElements)
void SetBuffer(const void *buffer)
RResult< void > VerifyChecksumIfEnabled() const
std::uint32_t GetBufferSize() const
void SetBufferSize(std::uint32_t bufferSize)