55 DistributionKey_t fDkey;
62static constexpr DistributionKey_t kDistributionKeyDefault = 0x5a3c69f0cafe4a11;
63static constexpr AttributeKey_t kAttributeKeyDefault = 0x4243544b53444229;
64static constexpr AttributeKey_t kAttributeKeyAnchor = 0x4243544b5344422a;
65static constexpr AttributeKey_t kAttributeKeyHeader = 0x4243544b5344422b;
66static constexpr AttributeKey_t kAttributeKeyFooter = 0x4243544b5344422c;
82 kDistributionKeyDefault, kAttributeKeyDefault};
87 std::string fPoolLabel;
89 std::string fContainerLabel;
95RDaosURI ParseDaosURI(std::string_view uri)
97 std::regex re(
"daos://([^/]+)/(.+)");
99 if (!std::regex_match(uri.data(),
m, re))
111struct RDaosContainerNTupleLocator {
114 std::optional<ROOT::Experimental::Internal::RDaosNTupleAnchor> fAnchor;
117 RDaosContainerNTupleLocator() =
default;
118 explicit RDaosContainerNTupleLocator(
const std::string &ntupleName) : fName(ntupleName), fIndex(
Hash(ntupleName)) {}
120 bool IsValid() {
return fAnchor.has_value() && fAnchor->fNBytesHeader; }
125 uint64_t
h = std::hash<std::string>{}(ntupleName);
127 auto seed =
static_cast<uint32_t
>(
h >> 32);
128 seed ^=
static_cast<uint32_t
>(
h & 0xffffffff) + 0x9e3779b9 + (seed << 6) + (seed >> 2);
130 return (hash == kReservedIndex) ? kReservedIndex + 1 : hash;
136 std::unique_ptr<unsigned char[]> buffer;
137 auto &anchor = fAnchor.emplace();
143 buffer = MakeUninitArray<unsigned char>(anchorSize);
144 if ((err = cont.
ReadSingleAkey(buffer.get(), anchorSize, oidMetadata, kDistributionKeyDefault,
145 kAttributeKeyAnchor, kCidAnchor))) {
149 anchor.Deserialize(buffer.get(), anchorSize).Unwrap();
151 builder.
SetVersion(anchor.fVersionEpoch, anchor.fVersionMajor, anchor.fVersionMinor, anchor.fVersionPatch);
158 static std::pair<RDaosContainerNTupleLocator, ROOT::Internal::RNTupleDescriptorBuilder>
164 auto &builder =
result.second;
166 loc.InitNTupleDescriptorBuilder(cont, builder);
177 if (buffer !=
nullptr) {
178 auto bytes =
reinterpret_cast<unsigned char *
>(buffer);
197 return R__FAIL(
"DAOS anchor too short");
199 auto bytes =
reinterpret_cast<const unsigned char *
>(buffer);
202 return R__FAIL(
"unsupported DAOS anchor version: " + std::to_string(fVersionAnchor));
216 return result.Unwrap() + 32;
230 static std::once_flag once;
231 std::call_once(once, []() {
233 <<
"Do not store real data with this version of RNTuple!";
245 auto args = ParseDaosURI(fURI);
246 auto pool = std::make_unique<RDaosPool>(args.fPoolLabel);
248 fDaosContainer = std::make_unique<RDaosContainer>(std::move(pool), args.fContainerLabel,
true);
249 fDaosContainer->SetDefaultObjectClass(fNTupleAnchor.fObjClass);
251 auto [locator,
_] = RDaosContainerNTupleLocator::LocateNTuple(*fDaosContainer, fNTupleName);
252 fNTupleIndex = locator.GetIndex();
254 auto zipBuffer = MakeUninitArray<unsigned char>(
length);
257 WriteNTupleHeader(zipBuffer.get(), szZipHeader,
length);
264 auto pageId = fPageId.fetch_add(1);
268 RDaosKey daosKey = GetPageDaosKey(fNTupleIndex, pageId);
269 fDaosContainer->WriteSingleAkey(sealedPage.
GetBuffer(), sealedPage.
GetBufferSize(), daosKey.fOid, daosKey.fDkey,
277 fCounters->fNPageCommitted.Inc();
283std::vector<ROOT::RNTupleLocator>
285 const std::vector<bool> &
mask)
288 std::vector<RNTupleLocator> locators;
289 auto nPages =
mask.size();
290 locators.reserve(nPages);
292 int64_t payloadSz = 0;
295 for (
auto &range : ranges) {
296 for (
auto sealedPageIt = range.fFirst; sealedPageIt != range.fLast; ++sealedPageIt) {
299 const auto pageId = fPageId.fetch_add(1);
304 RDaosKey daosKey = GetPageDaosKey(fNTupleIndex, pageId);
307 it->second.Insert(daosKey.fAkey, pageIov);
313 locators.push_back(locator);
318 fNBytesCurrentCluster += payloadSz;
322 if (
int err = fDaosContainer->WriteV(writeRequests))
326 fCounters->fNPageCommitted.Add(nPages);
327 fCounters->fSzWritePayload.Add(payloadSz);
334 return std::exchange(fNBytesCurrentCluster, 0);
341 auto bufPageListZip = MakeUninitArray<unsigned char>(
length);
345 auto offsetData = fClusterGroupId.fetch_add(1);
347 fDaosContainer->WriteSingleAkey(
348 bufPageListZip.get(),
350 daos_obj_id_t{kOidLowPageList, static_cast<decltype(daos_obj_id_t::hi)>(fNTupleIndex)},
351 kDistributionKeyDefault,
356 result.SetNBytesOnStorage(szPageListZip);
358 fCounters->fSzWritePayload.Add(
static_cast<int64_t
>(szPageListZip));
365 auto bufFooterZip = MakeUninitArray<unsigned char>(
length);
368 WriteNTupleFooter(bufFooterZip.get(), szFooterZip,
length);
377 fDaosContainer->WriteSingleAkey(
379 kDistributionKeyDefault, kAttributeKeyHeader);
380 fNTupleAnchor.fLenHeader = lenHeader;
381 fNTupleAnchor.fNBytesHeader = nbytes;
386 fDaosContainer->WriteSingleAkey(
388 kDistributionKeyDefault, kAttributeKeyFooter);
389 fNTupleAnchor.fLenFooter = lenFooter;
390 fNTupleAnchor.fNBytesFooter = nbytes;
396 auto buffer = MakeUninitArray<unsigned char>(ntplSize);
397 fNTupleAnchor.Serialize(buffer.get());
398 fDaosContainer->WriteSingleAkey(
399 buffer.get(), ntplSize,
daos_obj_id_t{kOidLowMetadata, static_cast<decltype(daos_obj_id_t::hi)>(fNTupleIndex)},
400 kDistributionKeyDefault, kAttributeKeyAnchor, kCidAnchor);
403std::unique_ptr<ROOT::Internal::RPageSink>
418 auto args = ParseDaosURI(uri);
419 auto pool = std::make_unique<RDaosPool>(args.fPoolLabel);
420 fDaosContainer = std::make_unique<RDaosContainer>(std::move(pool), args.fContainerLabel);
425 StopClusterPoolBackgroundThread();
430 RDaosContainerNTupleLocator ntupleLocator;
431 std::tie(ntupleLocator, fDescriptorBuilder) =
432 RDaosContainerNTupleLocator::LocateNTuple(*fDaosContainer, fNTupleName);
433 if (!ntupleLocator.IsValid()) {
435 R__FAIL(
"LoadStructureImpl: requested ntuple '" + fNTupleName +
"' is not present in DAOS container."));
437 fAnchor = *ntupleLocator.fAnchor;
438 fNTupleIndex = ntupleLocator.GetIndex();
440 fDaosContainer->SetDefaultObjectClass(fAnchor.fObjClass);
444 fAnchor.fNBytesHeader + fAnchor.fNBytesFooter + std::max(fAnchor.fLenHeader, fAnchor.fLenFooter);
445 fStructureBuffer.fBuffer = MakeUninitArray<unsigned char>(bufSize);
446 fStructureBuffer.fPtrHeader = fStructureBuffer.fBuffer.get();
447 fStructureBuffer.fPtrFooter = fStructureBuffer.fBuffer.get() + fAnchor.fNBytesHeader;
452 if ((err = fDaosContainer->ReadSingleAkey(fStructureBuffer.fPtrHeader, fAnchor.fNBytesHeader, oidMetadata,
453 kDistributionKeyDefault, kAttributeKeyHeader))) {
457 if ((err = fDaosContainer->ReadSingleAkey(fStructureBuffer.fPtrFooter, fAnchor.fNBytesFooter, oidMetadata,
458 kDistributionKeyDefault, kAttributeKeyFooter))) {
465 auto unzipBuf =
reinterpret_cast<unsigned char *
>(fStructureBuffer.fPtrFooter) + fAnchor.fNBytesFooter;
473 if (fDescriptorBuilder.GetDescriptor().GetName() != fNTupleName) {
475 throw ROOT::RException(
R__FAIL(
"LocateNTuple: ntuple name '" + fNTupleName +
"' unavailable in this container."));
478 return fDescriptorBuilder.MoveDescriptor();
482 unsigned char *buffer)
485 fDaosContainer->ReadSingleAkey(buffer, locator.
GetNBytesOnStorage(), oidPageList, kDistributionKeyDefault,
491 return fDaosContainer->GetDefaultObjectClass().ToString();
498 fDaosContainer->ReadSingleAkey(
const_cast<void *
>(sealedPage.
GetBuffer()), sealedPage.
GetBufferSize(), daosKey.fOid,
499 daosKey.fDkey, daosKey.fAkey);
504 auto clone = std::make_unique<RPageSourceDaos>(fNTupleName, fURI, fOptions);
505 clone->fAnchor = fAnchor;
506 clone->fNTupleIndex = fNTupleIndex;
507 if (!fAnchor.fObjClass.empty())
508 clone->fDaosContainer->SetDefaultObjectClass(fAnchor.fObjClass);
512std::vector<std::unique_ptr<RCluster>>
515 struct RDaosSealedPageLocator {
519 std::uint64_t fPageId = 0;
520 std::uint64_t fDataSize = 0;
521 std::uint64_t fBufferSize = 0;
528 auto fnPrepareSingleCluster = [&](
const RCluster::RKey &clusterKey,
531 std::vector<RDaosSealedPageLocator> onDiskPages;
533 unsigned clusterBufSz = 0, nPages = 0;
534 auto pageZeroMap = std::make_unique<ROOT::Internal::ROnDiskPageMap>();
536 clusterKey, *pageZeroMap,
539 const auto &pageLocator = pageInfo.
GetLocator();
541 const auto pageBufferSize = pageLocator.GetNBytesOnStorage() + pageInfo.
HasChecksum() * kNBytesPageChecksum;
542 onDiskPages.emplace_back(RDaosSealedPageLocator{clusterId, physicalColumnId, pageNo, pageId,
543 pageLocator.GetNBytesOnStorage(), pageBufferSize});
546 clusterBufSz += pageBufferSize;
549 auto clusterBuffer =
new unsigned char[clusterBufSz];
551 std::make_unique<ROOT::Internal::ROnDiskPageMapHeap>(std::unique_ptr<
unsigned char[]>(clusterBuffer));
554 for (
const auto &sealedLoc : onDiskPages) {
560 d_iov_set(&iov, clusterBuffer, sealedLoc.fBufferSize);
562 RDaosKey daosKey = GetPageDaosKey(fNTupleIndex, sealedLoc.fPageId);
565 itReq->second.Insert(daosKey.fAkey, iov);
567 clusterBuffer += sealedLoc.fBufferSize;
569 fCounters->fNPageRead.Add(nPages);
570 fCounters->fSzReadPayload.Add(clusterBufSz);
572 auto cluster = std::make_unique<RCluster>(clusterId);
573 cluster->Adopt(std::move(pageMap));
574 cluster->Adopt(std::move(pageZeroMap));
576 cluster->SetColumnAvailable(colId);
580 fCounters->fNClusterLoaded.Add(clusterKeys.size());
582 std::vector<std::unique_ptr<ROOT::Internal::RCluster>> clusters;
584 for (
auto key : clusterKeys) {
585 clusters.emplace_back(fnPrepareSingleCluster(key, readRequests));
590 if (
int err = fDaosContainer->ReadV(readRequests))
593 fCounters->fNReadV.Inc();
594 fCounters->fNRead.Add(readRequests.size());
599std::unique_ptr<ROOT::Internal::RPageSource>
#define R__FORWARD_ERROR(res)
Short-hand to return an RResult<T> in an error state (i.e. after checking)
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
#define R__LOG_WARNING(...)
Option_t Option_t TPoint TPoint const char GetTextMagnitude GetFillStyle GetLineColor GetLineWidth GetMarkerStyle GetTextAlign GetTextColor GetTextSize void data
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 mask
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
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 Int_t Int_t Window_t TString Int_t GCValues_t GetPrimarySelectionOwner GetDisplay GetScreen GetColormap GetNativeEvent const char const char dpyName wid window const char font_name cursor keysym reg const char only_if_exist regb h Point_t winding char text const char depth char const char Int_t count const char ColorStruct_t color const char Pixmap_t Pixmap_t PictureAttributes_t attr const char char ret_data h unsigned char height h length
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 Int_t Int_t Window_t TString Int_t GCValues_t GetPrimarySelectionOwner GetDisplay GetScreen GetColormap GetNativeEvent const char const char dpyName wid window const char font_name cursor keysym reg const char only_if_exist regb h Point_t winding char text const char depth char const char Int_t count const char ColorStruct_t color const char Pixmap_t Pixmap_t PictureAttributes_t attr const char char ret_data h unsigned char height h Atom_t Int_t ULong_t ULong_t bytes
UInt_t Hash(const TString &s)
Record wall time and CPU time between construction and destruction.
A RDaosContainer provides read/write access to objects in a given container.
RDaosObject::DistributionKey_t DistributionKey_t
std::unordered_map< ROidDkeyPair, RWOperation, ROidDkeyPair::Hash > MultiObjectRWOperation_t
int ReadSingleAkey(void *buffer, std::size_t length, daos_obj_id_t oid, DistributionKey_t dkey, AttributeKey_t akey, ObjClassId_t cid)
Read data from a single object attribute key to the given buffer.
RDaosObject::AttributeKey_t AttributeKey_t
std::unique_ptr< ROOT::Internal::RPageSink > CloneAsHidden(std::string_view name, const ROOT::RNTupleWriteOptions &opts) const final
Creates a new sink with the same underlying storage as this but writing to a different RNTuple named ...
std::vector< RNTupleLocator > CommitSealedPageVImpl(std::span< RPageStorage::RSealedPageGroup > ranges, const std::vector< bool > &mask) final
Vector commit of preprocessed pages.
void WriteNTupleFooter(const void *data, size_t nbytes, size_t lenFooter)
std::uint64_t StageClusterImpl() final
Returns the number of bytes written to storage (excluding metadata)
RNTupleLocator CommitClusterGroupImpl(unsigned char *serializedPageList, std::uint32_t length) final
Returns the locator of the page list envelope of the given buffer that contains the serialized page l...
void WriteNTupleHeader(const void *data, size_t nbytes, size_t lenHeader)
void InitImpl(unsigned char *serializedHeader, std::uint32_t length) final
~RPageSinkDaos() override
RNTupleLink CommitDatasetImpl() final
RPageSinkDaos(std::string_view ntupleName, std::string_view uri, const ROOT::RNTupleWriteOptions &options)
RNTupleLocator CommitSealedPageImpl(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage) final
ROOT::RNTupleDescriptor AttachImpl() final
LoadStructureImpl() has been called before AttachImpl() is called
std::string GetObjectClass() const
Return the object class used for user data OIDs in this ntuple.
std::unique_ptr< RPageSource > CloneImpl() const final
The cloned page source creates a new connection to the pool/container.
void LoadSealedPageImpl(const RNTupleLocator &locator, RSealedPage &sealedPage) final
std::vector< std::unique_ptr< ROOT::Internal::RCluster > > LoadClusters(std::span< ROOT::Internal::RCluster::RKey > clusterKeys) final
Populates all the pages of the given cluster ids and columns; it is possible that some columns do not...
void LoadPageListImpl(const RNTupleLocator &locator, unsigned char *buffer) final
void LoadStructureImpl() final
Fills fStructureBuffer with the compressed header and footer.
~RPageSourceDaos() override
std::unique_ptr< RPageSource > OpenWithDifferentAnchor(const ROOT::Internal::RNTupleLink &anchorLink, const ROOT::RNTupleReadOptions &options={}) final
Creates a new PageSource using the same underlying file as this but referring to a different RNTuple,...
std::unique_ptr< RDaosContainer > fDaosContainer
A container that stores object data (header/footer, pages, etc.)
RPageSourceDaos(std::string_view ntupleName, std::string_view uri, const ROOT::RNTupleReadOptions &options)
DAOS-specific user-tunable settings for storing ntuples.
const std::string & GetObjectClass() const
An in-memory subset of the packed and compressed pages of a cluster.
Helper class to compress data blocks in the ROOT compression frame format.
static std::size_t Zip(const void *from, std::size_t nbytes, int compression, void *to)
Returns the size of the compressed data, written into the provided output buffer.
Helper class to uncompress data blocks in the ROOT compression frame format.
static void Unzip(const void *from, size_t nbytes, size_t dataLen, void *to)
The nbytes parameter provides the size ls of the from buffer.
A helper class for piece-wise construction of an RNTupleDescriptor.
void SetVersion(std::uint16_t versionEpoch, std::uint16_t versionMajor, std::uint16_t versionMinor, std::uint16_t versionPatch)
void SetOnDiskHeaderSize(std::uint64_t size)
void AddToOnDiskFooterSize(std::uint64_t size)
The real footer size also include the page list envelopes.
A helper class for serializing and deserialization of the RNTuple binary format.
static RResult< std::uint32_t > DeserializeString(const void *buffer, std::uint64_t bufSize, std::string &val)
static std::uint32_t SerializeUInt32(std::uint32_t val, void *buffer)
static std::uint32_t DeserializeUInt32(const void *buffer, std::uint32_t &val)
static std::uint32_t SerializeUInt16(std::uint16_t val, void *buffer)
static RResult< void > DeserializeFooter(const void *buffer, std::uint64_t bufSize, ROOT::Internal::RNTupleDescriptorBuilder &descBuilder)
static std::uint32_t SerializeString(const std::string &val, void *buffer)
static std::uint32_t DeserializeUInt16(const void *buffer, std::uint16_t &val)
static RResult< void > DeserializeHeader(const void *buffer, std::uint64_t bufSize, ROOT::Internal::RNTupleDescriptorBuilder &descBuilder)
static std::uint32_t DeserializeUInt64(const void *buffer, std::uint64_t &val)
static std::uint32_t SerializeUInt64(std::uint64_t val, void *buffer)
A page as being stored on disk, that is packed and compressed.
Base class for a sink with a physical storage backend.
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSink.
Abstract interface to read data from an ntuple.
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSource.
Base class for all ROOT issued exceptions.
The on-storage metadata of an RNTuple.
RNTupleLocator payload that is common for object stores using 64bit location information.
std::uint64_t GetLocation() const
Generic information about the physical location of data.
std::uint64_t GetNBytesOnStorage() const
T GetPosition() const
Note that for GetPosition() / SetPosition(), the locator type must correspond (kTypeFile,...
void SetType(ELocatorType type)
void SetPosition(std::uint64_t position)
void SetNBytesOnStorage(std::uint64_t nBytesOnStorage)
Common user-tunable settings for reading RNTuples.
Common user-tunable settings for storing RNTuples.
The class is used as a return type for operations that can fail; wraps a value of type T or an RError...
const char * d_errstr(int rc)
static void d_iov_set(d_iov_t *iov, void *buf, size_t size)
uint16_t daos_oclass_id_t
std::uint32_t ntuple_index_t
ROOT::RLogChannel & NTupleLog()
Log channel for RNTuple diagnostics.
std::unique_ptr< T[]> MakeUninitArray(std::size_t size)
Make an array of default-initialized elements.
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.
A pair of <object ID, distribution key> that can be used to issue a fetch/update request for multiple...
Describes a read/write operation on multiple attribute keys under the same object ID and distribution...
Entry point for an RNTuple in a DAOS container.
std::uint32_t fNBytesFooter
The size of the compressed ntuple footer.
std::uint64_t fVersionAnchor
Allows for evolving the struct in future versions.
std::string fObjClass
The object class for user data OIDs, e.g. SX
std::uint16_t fVersionPatch
std::uint32_t Serialize(void *buffer) const
std::uint16_t fVersionEpoch
Version of the binary format supported by the writer.
std::uint16_t fVersionMinor
RResult< std::uint32_t > Deserialize(const void *buffer, std::uint32_t bufSize)
std::uint32_t fLenHeader
The size of the uncompressed ntuple header.
static std::uint32_t GetSize()
std::uint32_t fLenFooter
The size of the uncompressed ntuple footer.
std::uint16_t fVersionMajor
std::uint32_t fNBytesHeader
The size of the compressed ntuple header.
static constexpr std::size_t kOCNameMaxLength
This limit is currently not defined in any header and any call to daos_oclass_id2name() within DAOS u...
The identifiers that specifies the content of a (partial) cluster.
ROOT::DescriptorId_t fClusterId
ColumnSet_t fPhysicalColumnSet
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).
const void * GetBuffer() const
std::size_t GetBufferSize() const
std::size_t GetDataSize() const
Information about a single page in the context of a cluster's page range.
const RNTupleLocator & GetLocator() const