52 DistributionKey_t fDkey;
74template <EDaosMapping mapping>
81 static_cast<DistributionKey_t
>(
columnId),
static_cast<AttributeKey_t
>(
pageCount)};
91 std::string fPoolLabel;
93 std::string fContainerLabel;
101 std::regex re(
"daos://([^/]+)/(.+)");
103 if (!std::regex_match(uri.data(),
m, re))
112 auto position =
static_cast<uint32_t
>(address.
fLocation & 0xFFFFFFFF);
114 return {position,
offset};
121 uint64_t address = (position & 0xFFFFFFFF) | (
offset << 32);
132struct RDaosContainerNTupleLocator {
135 std::optional<ROOT::Experimental::Internal::RDaosNTupleAnchor> fAnchor;
138 RDaosContainerNTupleLocator() =
default;
141 bool IsValid() {
return fAnchor.has_value() && fAnchor->fNBytesHeader; }
148 auto seed =
static_cast<uint32_t
>(
h >> 32);
149 seed ^=
static_cast<uint32_t
>(
h & 0xffffffff) + 0x9e3779b9 + (seed << 6) + (seed >> 2);
151 return (
hash == kReservedIndex) ? kReservedIndex + 1 :
hash;
157 std::unique_ptr<unsigned char[]> buffer,
zipBuffer;
158 auto &
anchor = fAnchor.emplace();
164 buffer = std::make_unique<unsigned char[]>(
anchorSize);
173 R__FAIL(
"unsupported RNTuple epoch version: " + std::to_string(
anchor.fVersionEpoch)));
177 buffer = std::make_unique<unsigned char[]>(
anchor.fLenHeader);
187 buffer = std::make_unique<unsigned char[]>(
anchor.fLenFooter);
199 static std::pair<RDaosContainerNTupleLocator, ROOT::Experimental::Internal::RNTupleDescriptorBuilder>
206 auto &builder =
result.second;
208 if (
int err =
loc.InitNTupleDescriptorBuilder(
cont, builder); !err) {
212 R__FAIL(
"LocateNTuple: ntuple name '" +
ntupleName +
"' unavailable in this container."));
226 if (buffer !=
nullptr) {
227 auto bytes =
reinterpret_cast<unsigned char *
>(buffer);
246 return R__FAIL(
"DAOS anchor too short");
249 auto bytes =
reinterpret_cast<const unsigned char *
>(buffer);
252 return R__FAIL(
"unsupported DAOS anchor version: " + std::to_string(fVersionAnchor));
266 return result.Unwrap() + 32;
280 static std::once_flag
once;
281 std::call_once(
once, []() {
283 <<
"Do not store real data with this version of RNTuple!";
285 fCompressor = std::make_unique<RNTupleCompressor>();
304 auto pool = std::make_shared<RDaosPool>(args.fPoolLabel);
306 fDaosContainer = std::make_unique<RDaosContainer>(
pool, args.fContainerLabel,
true);
307 fDaosContainer->SetDefaultObjectClass(
oclass);
309 auto [
locator,
_] = RDaosContainerNTupleLocator::LocateNTuple(*fDaosContainer, fNTupleName);
310 fNTupleIndex =
locator.GetIndex();
328 fCounters->fSzZip.Add(
page.GetNBytes());
350 fCounters->fNPageCommitted.Inc();
351 fCounters->fSzWritePayload.Add(
sealedPage.GetBufferSize());
352 fNBytesCurrentCluster +=
sealedPage.GetBufferSize();
356std::vector<ROOT::Experimental::RNTupleLocator>
358 const std::vector<bool> &
mask)
361 std::vector<ROOT::Experimental::RNTupleLocator>
locators;
365 const uint32_t
maxCageSz = fCageSizeLimit;
366 const bool useCaging = fCageSizeLimit > 0;
375 for (
auto &
range : ranges) {
418 fCounters->fNPageCommitted.Add(
nPages);
419 fCounters->fSzWritePayload.Add(
payloadSz);
426 return std::exchange(fNBytesCurrentCluster, 0);
437 auto offsetData = fClusterGroupId.fetch_add(1);
438 fDaosContainer->WriteSingleAkey(
446 fCounters->fSzWritePayload.Add(
static_cast<int64_t
>(
szPageListZip));
462 fDaosContainer->WriteSingleAkey(
466 fNTupleAnchor.fNBytesHeader =
nbytes;
471 fDaosContainer->WriteSingleAkey(
475 fNTupleAnchor.fNBytesFooter =
nbytes;
481 auto buffer = std::make_unique<unsigned char[]>(
ntplSize);
482 fNTupleAnchor.Serialize(buffer.get());
483 fDaosContainer->WriteSingleAkey(
484 buffer.get(),
ntplSize,
daos_obj_id_t{kOidLowMetadata, static_cast<decltype(daos_obj_id_t::hi)>(fNTupleIndex)},
494 fClusterPool(std::make_unique<
RClusterPool>(*
this, options.GetClusterBunchSize()))
499 auto pool = std::make_shared<RDaosPool>(args.fPoolLabel);
508 std::unique_ptr<unsigned char[]> buffer,
zipBuffer;
510 auto [
locator,
descBuilder] = RDaosContainerNTupleLocator::LocateNTuple(*fDaosContainer, fNTupleName);
513 R__FAIL(
"Attach: requested ntuple '" + fNTupleName +
"' is not present in DAOS container."));
519 fDaosContainer->SetDefaultObjectClass(
oclass);
520 fNTupleIndex =
locator.GetIndex();
525 for (
const auto &
cgDesc : desc.GetClusterGroupIterable()) {
526 buffer = std::make_unique<unsigned char[]>(
cgDesc.GetPageListLength());
527 zipBuffer = std::make_unique<unsigned char[]>(
cgDesc.GetPageListLocator().fBytesOnStorage);
528 fDaosContainer->ReadSingleAkey(
532 cgDesc.GetPageListLength(), buffer.get());
542 return fDaosContainer->GetDefaultObjectClass().ToString();
582 fDaosContainer->ReadSingleAkey(
const_cast<void *
>(
sealedPage.GetBuffer()),
sealedPage.GetBufferSize(),
586 sealedPage.VerifyChecksumIfEnabled().ThrowOnError();
619 R__FAIL(
"accessing caged pages is only supported in conjunction with cluster cache"));
627 fCounters->fNPageRead.Inc();
628 fCounters->fNRead.Inc();
629 fCounters->fSzReadPayload.Add(
sealedPage.GetBufferSize());
632 if (!fCurrentCluster || (fCurrentCluster->GetId() !=
clusterId) || !fCurrentCluster->ContainsColumn(
columnId))
633 fCurrentCluster = fClusterPool->GetCluster(
clusterId, fActivePhysicalColumns.ToColumnSet());
641 auto onDiskPage = fCurrentCluster->GetOnDiskPage(key);
655 fCounters->fNPageUnsealed.Inc();
659std::unique_ptr<ROOT::Experimental::Internal::RPageSource>
663 return std::unique_ptr<RPageSourceDaos>(clone);
666std::vector<std::unique_ptr<ROOT::Experimental::Internal::RCluster>>
673 std::uint64_t fPosition = 0;
675 std::uint64_t fDataSize = 0;
676 std::uint64_t fBufferSize = 0;
688 std::unordered_map<std::uint32_t, std::vector<RDaosSealedPageLocator>>
onDiskPages;
691 auto pageZeroMap = std::make_unique<ROnDiskPageMap>();
697 uint32_t position,
offset;
699 auto [
itLoc,
_] =
onDiskPages.emplace(position, std::vector<RDaosSealedPageLocator>());
702 itLoc->second.push_back(
709 auto pageMap = std::make_unique<ROnDiskPageMapHeap>(std::unique_ptr<
unsigned char[]>(
clusterBuffer));
737 fCounters->fNPageRead.Add(
nPages);
748 fCounters->fNClusterLoaded.Add(
clusterKeys.size());
750 std::vector<std::unique_ptr<ROOT::Experimental::Internal::RCluster>>
clusters;
761 fCounters->fNReadV.Inc();
#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(...)
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.
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 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 offset
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)
Managed a set of clusters containing compressed and packed pages.
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
RDaosObject::AttributeKey_t AttributeKey_t
static Writer_t MakeMemCopyWriter(unsigned char *dest)
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 SetOnDiskHeaderSize(std::uint64_t size)
void AddToOnDiskFooterSize(std::uint64_t size)
The real footer size also include the page list envelopes.
const RNTupleDescriptor & GetDescriptor() const
A helper class for serializing and deserialization of the RNTuple binary format.
static std::uint32_t DeserializeUInt16(const void *buffer, std::uint16_t &val)
static RResult< void > DeserializeHeader(const void *buffer, std::uint64_t bufSize, RNTupleDescriptorBuilder &descBuilder)
static RResult< void > DeserializeFooter(const void *buffer, std::uint64_t bufSize, RNTupleDescriptorBuilder &descBuilder)
static std::uint32_t SerializeString(const std::string &val, void *buffer)
static std::uint32_t DeserializeUInt32(const void *buffer, std::uint32_t &val)
static std::uint32_t SerializeUInt64(std::uint64_t val, void *buffer)
static RResult< void > DeserializePageList(const void *buffer, std::uint64_t bufSize, DescriptorId_t clusterGroupId, RNTupleDescriptor &desc)
static std::uint32_t DeserializeUInt64(const void *buffer, std::uint64_t &val)
static RResult< std::uint32_t > DeserializeString(const void *buffer, std::uint64_t bufSize, std::string &val)
static std::uint32_t SerializeUInt16(std::uint16_t val, void *buffer)
static std::uint32_t SerializeUInt32(std::uint32_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.
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...
RPageSinkDaos(std::string_view ntupleName, std::string_view uri, const RNTupleWriteOptions &options)
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)
void WriteNTupleHeader(const void *data, size_t nbytes, size_t lenHeader)
void InitImpl(unsigned char *serializedHeader, std::uint32_t length) final
~RPageSinkDaos() override
RNTupleLocator CommitPageImpl(ColumnHandle_t columnHandle, const RPage &page) final
RNTupleLocator CommitSealedPageImpl(DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage) final
std::vector< RNTupleLocator > CommitSealedPageVImpl(std::span< RPageStorage::RSealedPageGroup > ranges, const std::vector< bool > &mask) final
Vector commit of preprocessed pages.
void CommitDatasetImpl() final
std::unique_ptr< RNTupleCompressor > fCompressor
Helper to zip pages and header/footer; includes a 16MB (kMAXZIPBUF) zip buffer.
Storage provider that reads ntuple pages from a DAOS container.
std::string GetObjectClass() const
Return the object class used for user data OIDs in this ntuple.
RPageRef LoadPageImpl(ColumnHandle_t columnHandle, const RClusterInfo &clusterInfo, ClusterSize_t::ValueType idxInCluster) final
std::vector< std::unique_ptr< RCluster > > LoadClusters(std::span< RCluster::RKey > clusterKeys) final
Populates all the pages of the given cluster ids and columns; it is possible that some columns do not...
RPageSourceDaos(std::string_view ntupleName, std::string_view uri, const RNTupleReadOptions &options)
~RPageSourceDaos() override
void LoadSealedPage(DescriptorId_t physicalColumnId, RClusterIndex clusterIndex, RSealedPage &sealedPage) final
Read the packed and compressed bytes of a page into the memory buffer provided by sealedPage.
RNTupleDescriptor AttachImpl() final
LoadStructureImpl() has been called before AttachImpl() is called
std::unique_ptr< RDaosContainer > fDaosContainer
A container that stores object data (header/footer, pages, etc.)
std::unique_ptr< RPageSource > CloneImpl() const final
The cloned page source creates a new connection to the pool/container.
Abstract interface to read data from an ntuple.
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSource.
Stores information about the cluster in which this page resides.
A page is a slice of a column that is mapped into memory.
static RPage MakePageZero(ColumnId_t columnId, ClusterSize_t::ValueType elementSize)
Make a 'zero' page for column columnId (that is comprised of 0x00 bytes only).
static const void * GetPageZeroBuffer()
Return a pointer to the page zero buffer used if there is no on-disk data for a particular deferred c...
Addresses a column element or field item relative to a particular cluster, instead of a global NTuple...
Base class for all ROOT issued exceptions.
The on-storage meta-data of an ntuple.
const std::string & GetName() const
Common user-tunable settings for reading ntuples.
DAOS-specific user-tunable settings for storing ntuples.
uint32_t GetMaxCageSize() const
const std::string & GetObjectClass() const
Common user-tunable settings for storing ntuples.
std::size_t GetMaxUnzippedPageSize() const
The class is used as a return type for operations that can fail; wraps a value of type T or an RError...
static constexpr std::uint16_t kVersionEpoch
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
RLogChannel & NTupleLog()
Log channel for RNTuple diagnostics.
std::uint64_t NTupleSize_t
Integer type long enough to hold the maximum number of entries in a column.
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
The identifiers that specifies the content of a (partial) cluster.
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.
RResult< std::uint32_t > Deserialize(const void *buffer, std::uint32_t bufSize)
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
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.
Wrap around a daos_oclass_id_t.
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...
On-disk pages within a page source are identified by the column and page number.
Summarizes cluster-level information that are necessary to load a certain page.
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
RNTupleLocator payload that is common for object stores using 64bit location information.
Generic information about the physical location of data.