61 std::unique_ptr<ROOT::RFieldBase>
CloneImpl(std::string_view )
const final
63 return std::make_unique<RRDFCardinalityField>();
98 *
static_cast<std::size_t *
>(to) =
size;
107 *
static_cast<std::size_t *
>(to) =
size;
121 std::unique_ptr<ROOT::RFieldBase>
CloneImpl(std::string_view)
const final
161 std::unique_ptr<RFieldBase::RValue>
fValue;
197 auto onDiskType =
source.GetSharedDescriptorGuard()->GetFieldDescriptor(
fField->GetOnDiskId()).GetTypeName();
198 std::string
msg =
"RNTupleDS: invalid type \"" +
fField->GetTypeName() +
"\" for column \"" +
201 throw std::runtime_error(
msg);
210 fValue = std::make_unique<RFieldBase::RValue>(
fField->CreateValue());
230 return fValue->GetPtr<
void>().get();
286 fieldDesc.GetTypeName().substr(0, 12) ==
"std::vector<" ||
fieldDesc.GetTypeName() ==
"");
316 f.SetOnDiskId(desc.
FindFieldId(
f.GetFieldName(),
f.GetParent()->GetOnDiskId()));
322 if (
info.fNRepetitions > 0) {
325 cardinalityField = std::make_unique<ROOT::Internal::RDF::RRDFCardinalityField>();
380 fPrincipalDescriptor =
pageSource->GetSharedDescriptorGuard()->Clone();
381 fStagingArea.emplace_back(std::move(
pageSource));
383 AddField(fPrincipalDescriptor,
"", fPrincipalDescriptor.GetFieldZeroId(),
384 std::vector<ROOT::RDF::RNTupleDS::RFieldInfo>());
395 static std::once_flag
flag;
396 std::call_once(
flag, []() {
407std::unique_ptr<ROOT::Internal::RPageSource> CreatePageSource(std::string_view
ntupleName, std::string_view fileName)
416 fFileNames = std::vector<std::string>{std::string{fileName}};
428 const std::pair<ULong64_t, ULong64_t> &
range)
445 std::distance(fColumnNames.begin(), std::find(fColumnNames.begin(), fColumnNames.end(),
fieldName));
450 return fProtoFields[
index].get();
465 return fld->GetTypeName() == requestedType;
474 throw std::runtime_error(
"RNTupleDS: Could not create field with type \"" +
requestedType +
475 "\" for column \"" + std::string(
fieldName) +
"\"");
485 return fProtoFields[
index].get();
488std::unique_ptr<ROOT::Detail::RDF::RColumnReaderBase>
496 fFieldId2QualifiedName[
field->GetOnDiskId()] = fPrincipalDescriptor.GetQualifiedFieldName(
field->GetOnDiskId());
497 for (
const auto &s : *
field) {
498 fFieldId2QualifiedName[s.GetOnDiskId()] = fPrincipalDescriptor.GetQualifiedFieldName(s.GetOnDiskId());
501 auto reader = std::make_unique<ROOT::Internal::RDF::RNTupleColumnReader>(
this,
field);
502 fActiveColumnReaders[
slot].emplace_back(
reader.get());
510 std::unique_lock lock(fMutexStaging);
511 fCvStaging.wait(lock, [
this] {
return fIsReadyForStaging || fStagingThreadShouldTerminate; });
512 if (fStagingThreadShouldTerminate)
517 fHasNextSources =
true;
518 fIsReadyForStaging =
false;
521 fCvStaging.notify_one();
527 const auto nFiles = fFileNames.empty() ? 1 : fFileNames.size();
529 for (
auto i = fNextFileIndex; (i <
nFiles) && ((i - fNextFileIndex) < fNSlots); ++i) {
531 if (fStagingThreadShouldTerminate)
534 if (fStagingArea[i]) {
538 fStagingArea[i] = CreatePageSource(fNTupleName, fFileNames[i]);
539 fStagingArea[i]->LoadStructure();
546 assert(fNextRanges.empty());
548 auto nFiles = fFileNames.empty() ? 1 : fFileNames.size();
557 while ((fNextRanges.size() < fNSlots) && (fNextFileIndex <
nFiles)) {
560 std::swap(fStagingArea[fNextFileIndex],
range.fSource);
562 if (!
range.fSource) {
565 range.fSource = CreatePageSource(fNTupleName, fFileNames[fNextFileIndex]);
567 range.fFileName = fFileNames[fNextFileIndex];
568 range.fSource->Attach();
575 fNextRanges.emplace_back(std::move(
range));
584 for (std::size_t i = 0; (fNextRanges.size() < fNSlots) && (fNextFileIndex <
nFiles); ++i) {
585 std::unique_ptr<ROOT::Internal::RPageSource>
source;
588 std::swap(fStagingArea[fNextFileIndex],
source);
591 source = CreatePageSource(fNTupleName, fFileNames[fNextFileIndex]);
616 unsigned int iSlot = 0;
632 range.fSource->SetEntryRange({start, end - start});
633 range.fFirstEntry = start;
634 range.fLastEntry = end;
635 fNextRanges.emplace_back(std::move(
range));
642 std::vector<std::pair<ULong64_t, ULong64_t>> ranges;
651 for (
auto r : fActiveColumnReaders[0]) {
652 r->Disconnect(
true );
658 if (fCurrentRanges.empty() || fSeenEntriesNoGlobalRange > 0) {
662 std::unique_lock lock(fMutexStaging);
663 fCvStaging.wait(lock, [
this] {
return fHasNextSources; });
666 if (fNextRanges.empty()) {
671 assert(fNextRanges.size() <= fNSlots);
673 fCurrentRanges.clear();
674 std::swap(fCurrentRanges, fNextRanges);
679 std::lock_guard
_(fMutexStaging);
680 fIsReadyForStaging =
true;
681 fHasNextSources =
false;
683 fCvStaging.notify_one();
690 fFirstEntry2RangeIdx.clear();
691 fOriginalRanges.clear();
695 for (std::size_t i = 0; i < fCurrentRanges.size(); ++i) {
699 if (fCurrentRanges[i].fFirstEntry == 0) {
705 auto start = fCurrentRanges[i].fFirstEntry + fSeenEntriesNoGlobalRange;
706 auto end = fCurrentRanges[i].fLastEntry + fSeenEntriesNoGlobalRange;
710 if (fGlobalEntryRange.has_value()) {
728 if (fGlobalEntryRange->first >= start && fGlobalEntryRange->second <= end) {
729 fOriginalRanges.emplace_back(start, end);
730 fFirstEntry2RangeIdx[fGlobalEntryRange->first] = i;
731 ranges.emplace_back(fGlobalEntryRange->first, fGlobalEntryRange->second);
736 else if (fGlobalEntryRange->second > end && fGlobalEntryRange->first < end) {
737 fOriginalRanges.emplace_back(start, end);
738 fFirstEntry2RangeIdx[fGlobalEntryRange->first] = i;
739 ranges.emplace_back(fGlobalEntryRange->first, end);
740 std::optional<std::pair<ULong64_t, ULong64_t>>
newvalues({end, fGlobalEntryRange->second});
744 else if (fGlobalEntryRange->second < start) {
749 else if (fGlobalEntryRange->first >= end) {
750 fOriginalRanges.emplace_back(start, end);
751 fFirstEntry2RangeIdx[start] = i;
752 ranges.emplace_back(start, start);
757 fFirstEntry2RangeIdx[start] = i;
758 fOriginalRanges.emplace_back(start, end);
759 ranges.emplace_back(start, end);
765 if ((fNSlots == 1) && (fCurrentRanges[0].fSource)) {
766 for (
auto r : fActiveColumnReaders[0]) {
767 r->Connect(*fCurrentRanges[0].fSource, fOriginalRanges[0].first);
778 fSlotsToRangeIdxs[0] = 0;
789 fSlotsToRangeIdxs[
slot * ROOT::Internal::RDF::CacheLineStep<std::size_t>()] =
idxRange;
791 for (
auto r : fActiveColumnReaders[
slot]) {
792 r->Connect(*fCurrentRanges[
idxRange].fSource,
802 for (
auto r : fActiveColumnReaders[
slot]) {
803 r->Disconnect(
true );
812 auto msg = std::string(
"RNTupleDS: There is no column with name \"") + std::string(
colName) +
"\"";
813 throw std::runtime_error(
msg);
817 return fColumnTypes[
index];
822 return std::find(fColumnNames.begin(), fColumnNames.end(),
colName) != fColumnNames.end();
827 fSeenEntriesNoGlobalRange = 0;
829 fIsReadyForStaging = fHasNextSources = fStagingThreadShouldTerminate =
false;
830 fThreadStaging = std::thread(&RNTupleDS::ExecStaging,
this);
831 assert(fNextRanges.empty());
833 if (fCurrentRanges.empty() || (fFileNames.size() > fNSlots)) {
836 std::lock_guard
_(fMutexStaging);
837 fIsReadyForStaging =
true;
839 fCvStaging.notify_one();
843 fNextFileIndex = std::max(fFileNames.size(), std::size_t(1));
849 for (
unsigned int i = 0; i < fNSlots; ++i) {
850 for (
auto r : fActiveColumnReaders[i]) {
851 r->Disconnect(
false );
855 std::lock_guard
_(fMutexStaging);
856 fStagingThreadShouldTerminate =
true;
858 fCvStaging.notify_one();
859 fThreadStaging.join();
862 if (fFileNames.size() > fNSlots) {
863 fCurrentRanges.clear();
865 fStagingArea.clear();
866 fStagingArea.resize(fFileNames.size());
875 fActiveColumnReaders.resize(fNSlots);
876 fSlotsToRangeIdxs.resize(fNSlots * ROOT::Internal::RDF::CacheLineStep<std::size_t>());
890 unsigned int slot,
const std::unordered_map<std::string, ROOT::RDF::Experimental::RSample *> &
sampleMap)
const
897 const auto &
rangeIdx = fSlotsToRangeIdxs.at(
slot * ROOT::Internal::RDF::CacheLineStep<std::size_t>());
900 if (!fCurrentRanges[
rangeIdx].fSource)
912 throw std::runtime_error(
"Full sample identifier '" +
ntupleID +
"' cannot be found in the available samples.");
921 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
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.
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
Namespace for new ROOT classes and functions.
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.