50 for (
unsigned i = 0; i < fIDs.size(); ++i) {
51 if (fIDs[i] == physicalColumnID) {
56 fIDs.emplace_back(physicalColumnID);
57 fRefCounters.emplace_back(1);
62 for (
unsigned i = 0; i < fIDs.size(); ++i) {
63 if (fIDs[i] == physicalColumnID) {
64 if (--fRefCounters[i] == 0) {
65 fIDs.erase(fIDs.begin() + i);
66 fRefCounters.erase(fRefCounters.begin() + i);
77 for (
const auto &
id : fIDs)
94 if (ntupleName.empty()) {
97 if (location.empty()) {
100 if (location.find(
"daos://") == 0)
102 return std::make_unique<RPageSourceDaos>(ntupleName, location, options);
107 return std::make_unique<RPageSourceFile>(ntupleName, location, options);
114 auto physicalId = GetSharedDescriptorGuard()->FindPhysicalColumnId(fieldId, column.
GetIndex());
116 fActivePhysicalColumns.Insert(physicalId);
122 fActivePhysicalColumns.Erase(columnHandle.
fPhysicalId);
127 return GetSharedDescriptorGuard()->GetNEntries();
132 return GetSharedDescriptorGuard()->GetNElements(columnHandle.
fPhysicalId);
144 UnzipClusterImpl(cluster);
156 auto pageBuffer = std::make_unique<unsigned char[]>(bytesPacked);
157 if (sealedPage.
fSize != bytesPacked) {
158 fDecompressor->Unzip(sealedPage.
fBuffer, sealedPage.
fSize, bytesPacked, pageBuffer.get());
163 memcpy(pageBuffer.get(), sealedPage.
fBuffer, bytesPacked);
167 auto unpackedBuffer =
new unsigned char[pageSize];
169 pageBuffer = std::unique_ptr<unsigned char []>(unpackedBuffer);
178 fCounters = std::unique_ptr<RCounters>(
new RCounters{
181 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"szReadPayload",
"B",
"volume read from storage (required)"),
182 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"szReadOverhead",
"B",
"volume read from storage (overhead)"),
185 "number of partial clusters preloaded from storage"),
186 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"nPageLoaded",
"",
"number of pages loaded from storage"),
187 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"nPagePopulated",
"",
"number of populated pages"),
188 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"timeWallRead",
"ns",
"wall clock time spent reading"),
189 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"timeWallUnzip",
"ns",
"wall clock time spent decompressing"),
192 "CPU time spent decompressing"),
193 *fMetrics.MakeCounter<
RNTupleCalcPerf*> (
"bwRead",
"MB/s",
"bandwidth compressed bytes read per second",
194 fMetrics, [](
const RNTupleMetrics &metrics) -> std::pair<bool, double> {
195 if (
const auto szReadPayload = metrics.GetLocalCounter(
"szReadPayload")) {
196 if (
const auto szReadOverhead = metrics.GetLocalCounter(
"szReadOverhead")) {
197 if (
const auto timeWallRead = metrics.GetLocalCounter(
"timeWallRead")) {
198 if (
auto walltime = timeWallRead->GetValueAsInt()) {
199 double payload = szReadPayload->GetValueAsInt();
200 double overhead = szReadOverhead->GetValueAsInt();
202 return {
true, (1000. * (payload + overhead) / walltime)};
210 *fMetrics.MakeCounter<
RNTupleCalcPerf*> (
"bwReadUnzip",
"MB/s",
"bandwidth uncompressed bytes read per second",
211 fMetrics, [](
const RNTupleMetrics &metrics) -> std::pair<bool, double> {
213 if (
const auto timeWallRead = metrics.
GetLocalCounter(
"timeWallRead")) {
214 if (
auto walltime = timeWallRead->GetValueAsInt()) {
217 return {
true, 1000. * unzip / walltime};
224 *fMetrics.MakeCounter<
RNTupleCalcPerf*> (
"bwUnzip",
"MB/s",
"decompression bandwidth of uncompressed bytes per second",
225 fMetrics, [](
const RNTupleMetrics &metrics) -> std::pair<bool, double> {
227 if (
const auto timeWallUnzip = metrics.
GetLocalCounter(
"timeWallUnzip")) {
228 if (
auto walltime = timeWallUnzip->GetValueAsInt()) {
231 return {
true, 1000. * unzip / walltime};
238 *fMetrics.MakeCounter<
RNTupleCalcPerf*> (
"rtReadEfficiency",
"",
"ratio of payload over all bytes read",
239 fMetrics, [](
const RNTupleMetrics &metrics) -> std::pair<bool, double> {
240 if (
const auto szReadPayload = metrics.
GetLocalCounter(
"szReadPayload")) {
241 if (
const auto szReadOverhead = metrics.
GetLocalCounter(
"szReadOverhead")) {
242 if (
auto payload = szReadPayload->GetValueAsInt()) {
244 return {
true, 1./(1. + (1. * szReadOverhead->GetValueAsInt()) / payload)};
251 *fMetrics.MakeCounter<
RNTupleCalcPerf*> (
"rtCompression",
"",
"ratio of compressed bytes / uncompressed bytes",
252 fMetrics, [](
const RNTupleMetrics &metrics) -> std::pair<bool, double> {
253 if (
const auto szReadPayload = metrics.
GetLocalCounter(
"szReadPayload")) {
255 if (
auto unzip = szUnzip->GetValueAsInt()) {
256 return {
true, (1. * szReadPayload->GetValueAsInt()) / unzip};
282 if (ntupleName.empty()) {
285 if (location.empty()) {
288 std::unique_ptr<ROOT::Experimental::Detail::RPageSink> realSink;
289 if (location.find(
"daos://") == 0) {
291 realSink = std::make_unique<RPageSinkDaos>(ntupleName, location, options);
296 realSink = std::make_unique<RPageSinkFile>(ntupleName, location, options);
300 return std::make_unique<RPageSinkBuf>(std::move(realSink));
307 auto columnId = fDescriptorBuilder.GetDescriptor().GetNPhysicalColumns();
308 fDescriptorBuilder.AddColumn(columnId, columnId, fieldId, column.
GetModel(), column.
GetIndex());
314 fDescriptorBuilder.SetNTuple(fNTupleName, model.
GetDescription());
315 const auto &descriptor = fDescriptorBuilder.GetDescriptor();
320 for (
auto &
f : fieldZero) {
321 auto fieldId = descriptor.GetNFields();
323 fDescriptorBuilder.AddFieldLink(
f.GetParent()->GetOnDiskId(), fieldId);
324 f.SetOnDiskId(fieldId);
325 f.ConnectPageSink(*
this);
330 auto fieldId = descriptor.GetNFields();
332 fDescriptorBuilder.AddFieldLink(
f.GetParent()->GetOnDiskId(), fieldId);
333 f.SetOnDiskId(fieldId);
335 for (
const auto &source : descriptor.GetColumnIterable(sourceFieldId)) {
336 auto targetId = descriptor.GetNLogicalColumns();
337 fDescriptorBuilder.AddColumn(targetId, source.GetLogicalId(), fieldId, source.GetModel(), source.GetIndex());
341 auto nColumns = descriptor.GetNPhysicalColumns();
348 fOpenColumnRanges.emplace_back(columnRange);
351 fOpenPageRanges.emplace_back(std::move(pageRange));
355 auto buffer = std::make_unique<unsigned char[]>(fSerializationContext.GetHeaderSize());
358 CreateImpl(model, buffer.get(), fSerializationContext.GetHeaderSize());
368 pageInfo.
fLocator = CommitPageImpl(columnHandle, page);
369 fOpenPageRanges.at(columnHandle.
fPhysicalId).fPageInfos.emplace_back(pageInfo);
376 fOpenColumnRanges.at(physicalColumnId).fNElements += sealedPage.
fNElements;
380 pageInfo.
fLocator = CommitSealedPageImpl(physicalColumnId, sealedPage);
381 fOpenPageRanges.at(physicalColumnId).fPageInfos.emplace_back(pageInfo);
384std::vector<ROOT::Experimental::RNTupleLocator>
387 std::vector<ROOT::Experimental::RNTupleLocator> locators;
388 for (
auto &range : ranges) {
389 for (
auto sealedPageIt = range.fFirst; sealedPageIt != range.fLast; ++sealedPageIt)
390 locators.push_back(CommitSealedPageImpl(range.fPhysicalColumnId, *sealedPageIt));
397 auto locators = CommitSealedPageVImpl(ranges);
400 for (
auto &range : ranges) {
401 for (
auto sealedPageIt = range.fFirst; sealedPageIt != range.fLast; ++sealedPageIt) {
402 fOpenColumnRanges.at(range.fPhysicalColumnId).fNElements += sealedPageIt->fNElements;
405 pageInfo.
fNElements = sealedPageIt->fNElements;
407 fOpenPageRanges.at(range.fPhysicalColumnId).fPageInfos.emplace_back(pageInfo);
414 auto nbytes = CommitClusterImpl(nEntries);
417 auto nEntriesInCluster =
ClusterSize_t(nEntries - fPrevClusterNEntries);
420 for (
unsigned int i = 0; i < fOpenColumnRanges.size(); ++i) {
423 std::swap(fullRange, fOpenPageRanges[i]);
425 fOpenColumnRanges[i].fCompressionSettings, fullRange);
426 fOpenColumnRanges[i].fFirstElementIndex += fOpenColumnRanges[i].fNElements;
427 fOpenColumnRanges[i].fNElements = 0;
429 fDescriptorBuilder.AddClusterWithDetails(clusterBuilder.
MoveDescriptor().Unwrap());
430 fPrevClusterNEntries = nEntries;
436 const auto &descriptor = fDescriptorBuilder.GetDescriptor();
438 const auto nClusters = descriptor.GetNClusters();
439 std::vector<DescriptorId_t> physClusterIDs;
440 for (
auto i = fNextClusterInGroup; i < nClusters; ++i) {
441 physClusterIDs.emplace_back(fSerializationContext.MapClusterId(i));
446 auto bufPageList = std::make_unique<unsigned char[]>(szPageList);
448 fSerializationContext);
450 const auto clusterGroupId = descriptor.GetNClusterGroups();
451 const auto locator = CommitClusterGroupImpl(bufPageList.get(), szPageList);
454 for (
auto i = fNextClusterInGroup; i < nClusters; ++i) {
457 fDescriptorBuilder.AddClusterGroup(std::move(cgBuilder));
458 fSerializationContext.MapClusterGroupId(clusterGroupId);
460 fNextClusterInGroup = nClusters;
465 const auto &descriptor = fDescriptorBuilder.GetDescriptor();
468 auto bufFooter = std::make_unique<unsigned char[]>(szFooter);
471 CommitDatasetImpl(bufFooter.get(), szFooter);
478 unsigned char *pageBuf =
reinterpret_cast<unsigned char *
>(page.
GetBuffer());
479 bool isAdoptedBuffer =
true;
484 pageBuf =
new unsigned char[packedBytes];
485 isAdoptedBuffer =
false;
488 auto zippedBytes = packedBytes;
490 if ((compressionSetting != 0) || !element.
IsMappable()) {
492 if (!isAdoptedBuffer)
494 pageBuf =
reinterpret_cast<unsigned char *
>(buf);
495 isAdoptedBuffer =
true;
508 return SealPage(page, element, compressionSetting, fCompressor->GetZipBuffer());
514 fCounters = std::unique_ptr<RCounters>(
new RCounters{
515 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"nPageCommitted",
"",
"number of pages committed to storage"),
516 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"szWritePayload",
"B",
"volume written for committed pages"),
518 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"timeWallWrite",
"ns",
"wall clock time spent writing"),
519 *fMetrics.MakeCounter<
RNTupleAtomicCounter*>(
"timeWallZip",
"ns",
"wall clock time spent compressing"),
522 "CPU time spent compressing")
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
Option_t Option_t TPoint TPoint const char GetTextMagnitude GetFillStyle GetLineColor GetLineWidth GetMarkerStyle GetTextAlign GetTextColor GetTextSize void char Point_t Rectangle_t WindowAttributes_t Float_t Float_t Float_t Int_t Int_t UInt_t UInt_t Rectangle_t result
An in-memory subset of the packed and compressed pages of a cluster.
std::unordered_set< DescriptorId_t > ColumnSet_t
virtual void Pack(void *destination, void *source, std::size_t count) const
If the on-storage layout and the in-memory layout differ, packing creates an on-disk page from an in-...
virtual bool IsMappable() const
Derived, typed classes tell whether the on-storage layout is bitwise identical to the memory layout.
virtual void Unpack(void *destination, void *source, std::size_t count) const
If the on-storage layout and the in-memory layout differ, unpacking creates a memory page from an on-...
std::size_t GetPackedSize(std::size_t nElements) const
std::size_t GetSize() const
const RColumnModel & GetModel() const
std::uint32_t GetIndex() const
void SetOnDiskId(DescriptorId_t id)
DescriptorId_t GetOnDiskId() const
A thread-safe integral performance counter.
A metric element that computes its floating point value from other counters.
std::int64_t GetValueAsInt() const override
size_t Zip(const void *from, size_t nbytes, int compression, Writer_t fnWriter)
Returns the size of the compressed data.
A collection of Counter objects with a name, a unit, and a description.
const RNTuplePerfCounter * GetLocalCounter(std::string_view name) const
Searches counters registered in this object only. Returns nullptr if name is not found.
An either thread-safe or non thread safe counter for CPU ticks.
void CommitDataset()
Finalize the current cluster and the entrire data set.
RSealedPage SealPage(const RPage &page, const RColumnElementBase &element, int compressionSetting)
Helper for streaming a page.
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSink.
void CommitSealedPageV(std::span< RPageStorage::RSealedPageGroup > ranges)
Write a vector of preprocessed pages to storage. The corresponding columns must have been added befor...
RPageSink(std::string_view ntupleName, const RNTupleWriteOptions &options)
void CommitPage(ColumnHandle_t columnHandle, const RPage &page)
Write a page to the storage. The column must have been added before.
static std::unique_ptr< RPageSink > Create(std::string_view ntupleName, std::string_view location, const RNTupleWriteOptions &options=RNTupleWriteOptions())
Guess the concrete derived page source from the file name (location)
std::uint64_t CommitCluster(NTupleSize_t nEntries)
Finalize the current cluster and create a new one for the following data.
void CommitClusterGroup()
Write out the page locations (page list envelope) for all the committed clusters since the last call ...
virtual std::vector< RNTupleLocator > CommitSealedPageVImpl(std::span< RPageStorage::RSealedPageGroup > ranges)
Vector commit of preprocessed pages.
void CommitSealedPage(DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage)
Write a preprocessed page to storage. The column must have been added before.
ColumnHandle_t AddColumn(DescriptorId_t fieldId, const RColumn &column) final
Register a new column.
RCluster::ColumnSet_t ToColumnSet() const
void Erase(DescriptorId_t physicalColumnID)
void Insert(DescriptorId_t physicalColumnID)
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSource.
std::unique_ptr< unsigned char[]> UnsealPage(const RSealedPage &sealedPage, const RColumnElementBase &element)
Helper for unstreaming a page.
NTupleSize_t GetNEntries()
void DropColumn(ColumnHandle_t columnHandle) override
Unregisters a column.
NTupleSize_t GetNElements(ColumnHandle_t columnHandle)
ColumnHandle_t AddColumn(DescriptorId_t fieldId, const RColumn &column) override
Register a new column.
static std::unique_ptr< RPageSource > Create(std::string_view ntupleName, std::string_view location, const RNTupleReadOptions &options=RNTupleReadOptions())
Guess the concrete derived page source from the file name (location)
RPageSource(std::string_view ntupleName, const RNTupleReadOptions &fOptions)
void UnzipCluster(RCluster *cluster)
Parallel decompression and unpacking of the pages in the given cluster.
ColumnId_t GetColumnId(ColumnHandle_t columnHandle)
Common functionality of an ntuple storage for both reading and writing.
RPageStorage(std::string_view name)
A page is a slice of a column that is mapped into memory.
std::uint32_t GetNBytes() const
The space taken by column elements in the buffer.
std::uint32_t GetNElements() const
static std::uint32_t SerializePageListV1(void *buffer, const RNTupleDescriptor &desc, std::span< DescriptorId_t > physClusterIDs, const RContext &context)
static RContext SerializeHeaderV1(void *buffer, const RNTupleDescriptor &desc)
static std::uint32_t SerializeFooterV1(void *buffer, const RNTupleDescriptor &desc, const RContext &context)
A helper class for piece-wise construction of an RClusterDescriptor.
RResult< void > CommitColumnRange(DescriptorId_t physicalId, std::uint64_t firstElementIndex, std::uint32_t compressionSettings, const RClusterDescriptor::RPageRange &pageRange)
RResult< RClusterDescriptor > MoveDescriptor()
Move out the full cluster descriptor including page locations.
A helper class for piece-wise construction of an RClusterGroupDescriptor.
RClusterGroupDescriptorBuilder & ClusterGroupId(DescriptorId_t clusterGroupId)
void AddCluster(DescriptorId_t clusterId)
RClusterGroupDescriptorBuilder & PageListLength(std::uint32_t pageListLength)
RClusterGroupDescriptorBuilder & PageListLocator(const RNTupleLocator &pageListLocator)
Base class for all ROOT issued exceptions.
static RFieldDescriptorBuilder FromField(const Detail::RFieldBase &field)
Make a new RFieldDescriptorBuilder based off a live NTuple field.
RFieldZero * GetFieldZero() const
const Detail::RFieldBase * GetSourceField(const Detail::RFieldBase *target) const
The RNTupleModel encapulates the schema of an ntuple.
std::string GetDescription() const
RFieldZero * GetFieldZero() const
const RProjectedFields & GetProjectedFields() const
Common user-tunable settings for reading ntuples.
Common user-tunable settings for storing ntuples.
bool GetUseBufferedWrite() const
basic_string_view< char > string_view
void swap(RDirectoryEntry &e1, RDirectoryEntry &e2) noexcept
std::uint64_t NTupleSize_t
Integer type long enough to hold the maximum number of entries in a column.
RClusterSize ClusterSize_t
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
std::int64_t ColumnId_t
Uniquely identifies a physical column within the scope of the current process, used to tag pages.
constexpr DescriptorId_t kInvalidDescriptorId
Default I/O performance counters that get registered in fMetrics.
Default I/O performance counters that get registered in fMetrics.
DescriptorId_t fPhysicalId
A sealed page contains the bytes of a page as written to storage (packed & compressed).
The window of element indexes of a particular column in a particular cluster.
std::int64_t fCompressionSettings
The usual format for ROOT compression settings (see Compression.h).
NTupleSize_t fFirstElementIndex
A 64bit element index.
ClusterSize_t fNElements
The number of column elements in the cluster.
DescriptorId_t fPhysicalColumnId