61 std::unique_ptr<ROOT::RFieldBase>
CloneImpl(std::string_view )
const final
63 return std::make_unique<RRDFCardinalityField>();
101 *
static_cast<std::size_t *
>(to) =
size;
110 *
static_cast<std::size_t *
>(to) =
size;
124 std::unique_ptr<ROOT::RFieldBase>
CloneImpl(std::string_view)
const final
167 std::unique_ptr<RFieldBase::RValue>
fValue;
203 auto onDiskType =
source.GetSharedDescriptorGuard()->GetFieldDescriptor(
fField->GetOnDiskId()).GetTypeName();
204 std::string
msg =
"RNTupleDS: invalid type \"" +
fField->GetTypeName() +
"\" for column \"" +
207 throw std::runtime_error(
msg);
216 fValue = std::make_unique<RFieldBase::RValue>(
fField->CreateValue());
236 return fValue->GetPtr<
void>().get();
292 fieldDesc.GetTypeName().substr(0, 12) ==
"std::vector<" ||
fieldDesc.GetTypeName() ==
"");
322 f.SetOnDiskId(desc.
FindFieldId(
f.GetFieldName(),
f.GetParent()->GetOnDiskId()));
328 if (
info.fNRepetitions > 0) {
331 cardinalityField = std::make_unique<ROOT::Internal::RDF::RRDFCardinalityField>();
386 fPrincipalDescriptor =
pageSource->GetSharedDescriptorGuard()->Clone();
387 fStagingArea.emplace_back(std::move(
pageSource));
389 AddField(fPrincipalDescriptor,
"", fPrincipalDescriptor.GetFieldZeroId(),
390 std::vector<ROOT::RDF::RNTupleDS::RFieldInfo>());
401 static std::once_flag
flag;
402 std::call_once(
flag, []() {
413std::unique_ptr<ROOT::Internal::RPageSource> CreatePageSource(std::string_view
ntupleName, std::string_view fileName)
422 fFileNames = std::vector<std::string>{std::string{fileName}};
434 const std::pair<ULong64_t, ULong64_t> &
range)
451 std::distance(fColumnNames.begin(), std::find(fColumnNames.begin(), fColumnNames.end(),
fieldName));
456 return fProtoFields[
index].get();
471 return fld->GetTypeName() == requestedType;
480 throw std::runtime_error(
"RNTupleDS: Could not create field with type \"" +
requestedType +
481 "\" for column \"" + std::string(
fieldName) +
"\"");
491 return fProtoFields[
index].get();
494std::unique_ptr<ROOT::Detail::RDF::RColumnReaderBase>
502 fFieldId2QualifiedName[
field->GetOnDiskId()] = fPrincipalDescriptor.GetQualifiedFieldName(
field->GetOnDiskId());
503 for (
const auto &s : *
field) {
504 fFieldId2QualifiedName[s.GetOnDiskId()] = fPrincipalDescriptor.GetQualifiedFieldName(s.GetOnDiskId());
507 auto reader = std::make_unique<ROOT::Internal::RDF::RNTupleColumnReader>(
this,
field);
508 fActiveColumnReaders[
slot].emplace_back(
reader.get());
516 std::unique_lock lock(fMutexStaging);
517 fCvStaging.wait(lock, [
this] {
return fIsReadyForStaging || fStagingThreadShouldTerminate; });
518 if (fStagingThreadShouldTerminate)
523 fHasNextSources =
true;
524 fIsReadyForStaging =
false;
527 fCvStaging.notify_one();
533 const auto nFiles = fFileNames.empty() ? 1 : fFileNames.size();
535 for (
auto i = fNextFileIndex; (i <
nFiles) && ((i - fNextFileIndex) < fNSlots); ++i) {
537 if (fStagingThreadShouldTerminate)
540 if (fStagingArea[i]) {
544 fStagingArea[i] = CreatePageSource(fNTupleName, fFileNames[i]);
545 fStagingArea[i]->LoadStructure();
552 assert(fNextRanges.empty());
554 auto nFiles = fFileNames.empty() ? 1 : fFileNames.size();
563 while ((fNextRanges.size() < fNSlots) && (fNextFileIndex <
nFiles)) {
566 std::swap(fStagingArea[fNextFileIndex],
range.fSource);
568 if (!
range.fSource) {
571 range.fSource = CreatePageSource(fNTupleName, fFileNames[fNextFileIndex]);
573 range.fFileName = fFileNames[fNextFileIndex];
574 range.fSource->Attach();
581 fNextRanges.emplace_back(std::move(
range));
590 for (std::size_t i = 0; (fNextRanges.size() < fNSlots) && (fNextFileIndex <
nFiles); ++i) {
591 std::unique_ptr<ROOT::Internal::RPageSource>
source;
594 std::swap(fStagingArea[fNextFileIndex],
source);
597 source = CreatePageSource(fNTupleName, fFileNames[fNextFileIndex]);
622 unsigned int iSlot = 0;
638 range.fSource->SetEntryRange({start, end - start});
639 range.fFirstEntry = start;
640 range.fLastEntry = end;
641 fNextRanges.emplace_back(std::move(
range));
648 std::vector<std::pair<ULong64_t, ULong64_t>> ranges;
657 for (
auto r : fActiveColumnReaders[0]) {
658 r->Disconnect(
true );
664 if (fCurrentRanges.empty() || fSeenEntriesNoGlobalRange > 0) {
668 std::unique_lock lock(fMutexStaging);
669 fCvStaging.wait(lock, [
this] {
return fHasNextSources; });
672 if (fNextRanges.empty()) {
677 assert(fNextRanges.size() <= fNSlots);
679 fCurrentRanges.clear();
680 std::swap(fCurrentRanges, fNextRanges);
685 std::lock_guard
_(fMutexStaging);
686 fIsReadyForStaging =
true;
687 fHasNextSources =
false;
689 fCvStaging.notify_one();
696 fFirstEntry2RangeIdx.clear();
697 fOriginalRanges.clear();
701 for (std::size_t i = 0; i < fCurrentRanges.size(); ++i) {
705 if (fCurrentRanges[i].fFirstEntry == 0) {
711 auto start = fCurrentRanges[i].fFirstEntry + fSeenEntriesNoGlobalRange;
712 auto end = fCurrentRanges[i].fLastEntry + fSeenEntriesNoGlobalRange;
716 if (fGlobalEntryRange.has_value()) {
734 if (fGlobalEntryRange->first >= start && fGlobalEntryRange->second <= end) {
735 fOriginalRanges.emplace_back(start, end);
736 fFirstEntry2RangeIdx[fGlobalEntryRange->first] = i;
737 ranges.emplace_back(fGlobalEntryRange->first, fGlobalEntryRange->second);
742 else if (fGlobalEntryRange->second > end && fGlobalEntryRange->first < end) {
743 fOriginalRanges.emplace_back(start, end);
744 fFirstEntry2RangeIdx[fGlobalEntryRange->first] = i;
745 ranges.emplace_back(fGlobalEntryRange->first, end);
746 std::optional<std::pair<ULong64_t, ULong64_t>>
newvalues({end, fGlobalEntryRange->second});
750 else if (fGlobalEntryRange->second < start) {
755 else if (fGlobalEntryRange->first >= end) {
756 fOriginalRanges.emplace_back(start, end);
757 fFirstEntry2RangeIdx[start] = i;
758 ranges.emplace_back(start, start);
763 fFirstEntry2RangeIdx[start] = i;
764 fOriginalRanges.emplace_back(start, end);
765 ranges.emplace_back(start, end);
771 if ((fNSlots == 1) && (fCurrentRanges[0].fSource)) {
772 for (
auto r : fActiveColumnReaders[0]) {
773 r->Connect(*fCurrentRanges[0].fSource, fOriginalRanges[0].first);
784 fSlotsToRangeIdxs[0] = 0;
795 fSlotsToRangeIdxs[
slot * ROOT::Internal::RDF::CacheLineStep<std::size_t>()] =
idxRange;
797 for (
auto r : fActiveColumnReaders[
slot]) {
798 r->Connect(*fCurrentRanges[
idxRange].fSource,
808 for (
auto r : fActiveColumnReaders[
slot]) {
809 r->Disconnect(
true );
818 auto msg = std::string(
"RNTupleDS: There is no column with name \"") + std::string(
colName) +
"\"";
819 throw std::runtime_error(
msg);
823 return fColumnTypes[
index];
828 return std::find(fColumnNames.begin(), fColumnNames.end(),
colName) != fColumnNames.end();
833 fSeenEntriesNoGlobalRange = 0;
835 fIsReadyForStaging = fHasNextSources = fStagingThreadShouldTerminate =
false;
836 fThreadStaging = std::thread(&RNTupleDS::ExecStaging,
this);
837 assert(fNextRanges.empty());
839 if (fCurrentRanges.empty() || (fFileNames.size() > fNSlots)) {
842 std::lock_guard
_(fMutexStaging);
843 fIsReadyForStaging =
true;
845 fCvStaging.notify_one();
849 fNextFileIndex = std::max(fFileNames.size(), std::size_t(1));
855 for (
unsigned int i = 0; i < fNSlots; ++i) {
856 for (
auto r : fActiveColumnReaders[i]) {
857 r->Disconnect(
false );
861 std::lock_guard
_(fMutexStaging);
862 fStagingThreadShouldTerminate =
true;
864 fCvStaging.notify_one();
865 fThreadStaging.join();
868 if (fFileNames.size() > fNSlots) {
869 fCurrentRanges.clear();
871 fStagingArea.clear();
872 fStagingArea.resize(fFileNames.size());
881 fActiveColumnReaders.resize(fNSlots);
882 fSlotsToRangeIdxs.resize(fNSlots * ROOT::Internal::RDF::CacheLineStep<std::size_t>());
896 unsigned int slot,
const std::unordered_map<std::string, ROOT::RDF::Experimental::RSample *> &
sampleMap)
const
903 const auto &
rangeIdx = fSlotsToRangeIdxs.at(
slot * ROOT::Internal::RDF::CacheLineStep<std::size_t>());
906 if (!fCurrentRanges[
rangeIdx].fSource)
918 throw std::runtime_error(
"Full sample identifier '" +
ntupleID +
"' cannot be found in the available samples.");
927 const std::pair<ULong64_t, ULong64_t> &
range)
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
size_t size(const MatrixT &matrix)
retrieve the size of a square matrix
long long Long64_t
Portable signed long integer 8 bytes.
unsigned long long ULong64_t
Portable unsigned long integer 8 bytes.
ROOT::Detail::TRangeCast< T, true > TRangeDynCast
TRangeDynCast is an adapter class that allows the typed iteration through a TCollection.
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 r
Option_t Option_t TPoint TPoint const char GetTextMagnitude GetFillStyle GetLineColor GetLineWidth GetMarkerStyle GetTextAlign GetTextColor GetTextSize void char Point_t Rectangle_t WindowAttributes_t index
R__EXTERN TSystem * gSystem
void GetCollectionInfo(const ROOT::NTupleSize_t globalIndex, RNTupleLocalIndex *collectionStart, ROOT::NTupleSize_t *collectionSize)
For offset columns only, look at the two adjacent values that define a collection's coordinates.
An artificial field that provides the size of a fixed-size array.
void GenerateColumns(const ROOT::RNTupleDescriptor &) final
Implementations in derived classes should create the backing columns corresponding to the field type ...
std::unique_ptr< ROOT::RFieldBase > CloneImpl(std::string_view) const final
Called by Clone(), which additionally copies the on-disk ID.
void ReadGlobalImpl(ROOT::NTupleSize_t, void *to) final
RArraySizeField(const RArraySizeField &other)=delete
RArraySizeField(RArraySizeField &&other)=default
RArraySizeField & operator=(RArraySizeField &&other)=default
void ReadInClusterImpl(RNTupleLocalIndex, void *to) final
void ReconcileOnDiskField(const RNTupleDescriptor &) final
For non-artificial fields, check compatibility of the in-memory field and the on-disk field.
std::size_t GetValueSize() const final
The number of bytes taken by a value of the appropriate type.
RArraySizeField(std::size_t arrayLength)
void ConstructValue(void *where) const final
Constructs value in a given location of size at least GetValueSize(). Called by the base class' Creat...
~RArraySizeField() final=default
RArraySizeField & operator=(const RArraySizeField &other)=delete
std::size_t GetAlignment() const final
As a rule of thumb, the alignment is equal to the size of the type.
void GenerateColumns() final
Implementations in derived classes should create the backing columns corresponding to the field type ...
Every RDF column is represented by exactly one RNTuple field.
void * GetImpl(Long64_t entry) final
void Connect(RPageSource &source, Long64_t entryOffset)
Connect the field and its subfields to the page source.
RNTupleColumnReader(RNTupleDS *ds, RFieldBase *protoField)
std::unique_ptr< RFieldBase::RValue > fValue
The memory location used to read from fField.
std::unique_ptr< RFieldBase > fField
The field backing the RDF column.
Long64_t fEntryOffset
For chains, the logical entry and the physical entry in any particular file can be different.
std::shared_ptr< void > fValuePtr
Used to reuse the object created by fValue when reconnecting sources.
RNTupleDS * fDataSource
The data source that owns this column reader.
~RNTupleColumnReader() override=default
RFieldBase * fProtoField
The prototype field from which fField is cloned.
Long64_t fLastEntry
Last entry number that was read.
void Disconnect(bool keepValue)
An artificial field that transforms an RNTuple column that contains the offset of collections into co...
RRDFCardinalityField(RRDFCardinalityField &&other)=default
size_t GetAlignment() const final
As a rule of thumb, the alignment is equal to the size of the type.
void GenerateColumns() final
Implementations in derived classes should create the backing columns corresponding to the field type ...
void ReadGlobalImpl(ROOT::NTupleSize_t globalIndex, void *to) final
Get the number of elements of the collection identified by globalIndex.
void ReconcileOnDiskField(const RNTupleDescriptor &) final
For non-artificial fields, check compatibility of the in-memory field and the on-disk field.
RRDFCardinalityField & operator=(RRDFCardinalityField &&other)=default
void GenerateColumns(const ROOT::RNTupleDescriptor &desc) final
Implementations in derived classes should create the backing columns corresponding to the field type ...
~RRDFCardinalityField() override=default
const RColumnRepresentations & GetColumnRepresentations() const final
Implementations in derived classes should return a static RColumnRepresentations object.
size_t GetValueSize() const final
The number of bytes taken by a value of the appropriate type.
void ReadInClusterImpl(ROOT::RNTupleLocalIndex localIndex, void *to) final
Get the number of elements of the collection identified by clusterIndex.
std::unique_ptr< ROOT::RFieldBase > CloneImpl(std::string_view) const final
Called by Clone(), which additionally copies the on-disk ID.
void ConstructValue(void *where) const final
Constructs value in a given location of size at least GetValueSize(). Called by the base class' Creat...
static void SetClusterBunchSize(RNTupleReadOptions &options, unsigned int val)
Abstract interface to read data from an ntuple.
static std::unique_ptr< RPageSource > Create(std::string_view ntupleName, std::string_view location, const ROOT::RNTupleReadOptions &options=ROOT::RNTupleReadOptions())
Guess the concrete derived page source from the file name (location)
std::vector< void * > Record_t
std::optional< std::pair< ULong64_t, ULong64_t > > fGlobalEntryRange
The RDataSource implementation for RNTuple.
void AddField(const ROOT::RNTupleDescriptor &desc, std::string_view colName, ROOT::DescriptorId_t fieldId, std::vector< RFieldInfo > fieldInfos, bool convertToRVec=true)
Provides the RDF column "colName" given the field identified by fieldID.
std::vector< std::pair< ULong64_t, ULong64_t > > GetEntryRanges() final
Return ranges of entries to distribute to tasks.
void ExecStaging()
The main function of the fThreadStaging background thread.
std::vector< std::unique_ptr< ROOT::Internal::RPageSource > > fStagingArea
The staging area is relevant for chains of files, i.e.
std::unique_ptr< ROOT::Detail::RDF::RColumnReaderBase > GetColumnReaders(unsigned int, std::string_view, const std::type_info &) final
If the other GetColumnReaders overload returns an empty vector, this overload will be called instead.
std::vector< std::unique_ptr< ROOT::RFieldBase > > fProtoFields
We prepare a prototype field for every column.
void SetNSlots(unsigned int nSlots) final
Inform RDataSource of the number of processing slots (i.e.
ROOT::RFieldBase * GetFieldWithTypeChecks(std::string_view fieldName, const std::type_info &tid)
std::vector< std::string > fFileNames
void InitSlot(unsigned int slot, ULong64_t firstEntry) final
Convenience method called at the start of the data processing associated to a slot.
RNTupleDS(std::unique_ptr< ROOT::Internal::RPageSource > pageSource)
std::string GetTypeName(std::string_view colName) const final
Type of a column as a string, e.g.
std::unordered_map< ROOT::DescriptorId_t, std::string > fFieldId2QualifiedName
Connects the IDs of active proto fields and their subfields to their fully qualified name (a....
std::string fNTupleName
The data source may be constructed with an ntuple name and a list of files.
void PrepareNextRanges()
Populates fNextRanges with the next set of entry ranges.
void StageNextSources()
Starting from fNextFileIndex, opens the next fNSlots files.
void Finalize() final
Convenience method called after concluding an event-loop.
std::vector< std::string > fColumnTypes
void Initialize() final
Convenience method called before starting an event-loop.
std::vector< std::string > fColumnNames
bool HasColumn(std::string_view colName) const final
Checks if the dataset has a certain column.
Record_t GetColumnReadersImpl(std::string_view name, const std::type_info &) final
type-erased vector of pointers to pointers to column values - one per slot
void FinalizeSlot(unsigned int slot) final
Convenience method called at the end of the data processing associated to a slot.
This type represents a sample identifier, to be used in conjunction with RDataFrame features such as ...
ROOT's RDataFrame offers a modern, high-level interface for analysis of data stored in TTree ,...
Base class for all ROOT issued exceptions.
The list of column representations a field can have.
A field translates read and write calls from/to underlying columns to/from tree values.
ROOT::Internal::RColumn * fPrincipalColumn
All fields that have columns have a distinct main column.
RConstSchemaIterator cbegin() const
const std::string & GetFieldName() const
static RResult< std::unique_ptr< RFieldBase > > Create(const std::string &fieldName, const std::string &typeName, const ROOT::RCreateFieldOptions &options, const ROOT::RNTupleDescriptor *desc, ROOT::DescriptorId_t fieldId)
Factory method to resurrect a field from the stored on-disk type information.
ROOT::DescriptorId_t GetOnDiskId() const
std::unique_ptr< RFieldBase > Clone(std::string_view newName) const
Copies the field and its subfields using a possibly new name and a new, unconnected set of columns.
The on-storage metadata of an RNTuple.
RFieldDescriptorIterable GetFieldIterable(const RFieldDescriptor &fieldDesc) const
const RFieldDescriptor & GetFieldDescriptor(ROOT::DescriptorId_t fieldId) const
ROOT::DescriptorId_t FindFieldId(std::string_view fieldName, ROOT::DescriptorId_t parentId) const
Addresses a column element or field item relative to a particular cluster, instead of a global NTuple...
Common user-tunable settings for reading RNTuples.
const_iterator begin() const
const_iterator end() const
virtual const char * Getenv(const char *env)
Get environment variable.
std::string TypeID2TypeName(const std::type_info &id)
Returns the name of a type starting from its type_info An empty string is returned in case of failure...
ROOT::RDataFrame FromRNTuple(std::string_view ntupleName, const std::vector< std::string > &fileNames, const std::pair< ULong64_t, ULong64_t > &range)
Internal overload of the function that allows passing a range of entries.
std::pair< std::vector< ROOT::Internal::RNTupleClusterBoundaries >, ROOT::NTupleSize_t > GetClustersAndEntries(std::string_view ntupleName, std::string_view location)
Retrieves the cluster boundaries and the number of entries for the input RNTuple.
void CallConnectPageSourceOnField(RFieldBase &, ROOT::Internal::RPageSource &)
std::vector< ROOT::Internal::RNTupleClusterBoundaries > GetClusterBoundaries(const RNTupleDescriptor &desc)
Return the cluster boundaries for each cluster in this RNTuple.
std::string GetRenormalizedTypeName(const std::string &metaNormalizedName)
Given a type name normalized by ROOT meta, renormalize it for RNTuple. E.g., insert std::prefix.
RDataFrame FromRNTuple(std::string_view ntupleName, std::string_view fileName)
std::vector< std::string > ColumnNames_t
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.
ENTupleStructure
The fields in the RNTuple data model tree can carry different structural information about the type s...
Tag to let data sources use the native data type when creating a column reader.
The PrepareNextRanges() method populates the fNextRanges list with REntryRangeDS records.