Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions tree/dataframe/inc/ROOT/RNTupleProcessorDS.hxx
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,11 @@ class RNTupleProcessorDS final : public ROOT::RDF::RDataSource {
/// be added as `ROOT::VecOps::RVec` (e.g., `std::set<Jet> will be added as a `std::set`, but `Jet.[pt|eta] will be
/// added as `ROOT::VecOps::RVec<float>).
void AddField(const ROOT::RFieldBase &field, std::string_view colName,
Internal::RNTupleProcessorProvenance procProvenance, std::deque<RFieldInfo> &fieldInfos,
ROOT::Internal::RNTupleProcessorProvenance procProvenance, std::deque<RFieldInfo> &fieldInfos,
bool convertToRVec = true);

public:
RNTupleProcessorDS(std::unique_ptr<ROOT::Experimental::RNTupleProcessor> processor);
RNTupleProcessorDS(std::unique_ptr<ROOT::RNTupleProcessor> processor);
RNTupleProcessorDS(const RNTupleProcessorDS &) = delete;
RNTupleProcessorDS &operator=(const RNTupleProcessorDS &) = delete;
RNTupleProcessorDS(RNTupleProcessorDS &&) = delete;
Expand Down Expand Up @@ -132,7 +132,7 @@ protected:
} // namespace ROOT::Experimental::Internal::RDF

namespace ROOT::Experimental::RDF {
RDataFrame FromRNTupleProcessor(std::unique_ptr<ROOT::Experimental::RNTupleProcessor> processor);
RDataFrame FromRNTupleProcessor(std::unique_ptr<ROOT::RNTupleProcessor> processor);
} // namespace ROOT::Experimental::RDF

#endif // ROOT_RNTupleProcessorDS
17 changes: 8 additions & 9 deletions tree/dataframe/src/RNTupleProcessorDS.cxx
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ RNTupleProcessorDS::~RNTupleProcessorDS() = default;

// TODO: see if common parts of this and RNTupleDS::AddField can be factored out.
void RNTupleProcessorDS::AddField(const ROOT::RFieldBase &field, std::string_view colName,
Internal::RNTupleProcessorProvenance procProvenance,
ROOT::Internal::RNTupleProcessorProvenance procProvenance,
std::deque<RNTupleProcessorDS::RFieldInfo> &fieldInfos, bool convertToRVec)
{
// As an example for the mapping of RNTuple fields to RDF columns, let's consider an RNTuple
Expand Down Expand Up @@ -208,7 +208,7 @@ void RNTupleProcessorDS::AddField(const ROOT::RFieldBase &field, std::string_vie
fColumnNames.emplace_back(cardinalityFieldName);
fColumnTypes.emplace_back(cardinalityField->GetTypeName());
fProcessor->AddFieldToEntry(std::move(cardinalityField), cardinalityFieldName, nullptr,
Internal::RNTupleProcessorProvenance());
ROOT::Internal::RNTupleProcessorProvenance());
}

fieldInfos.emplace_back(field.GetOnDiskId(), field.GetFieldName(), field.GetTypeName(), nRepetitions);
Expand All @@ -220,19 +220,19 @@ void RNTupleProcessorDS::AddField(const ROOT::RFieldBase &field, std::string_vie
// Add nested fields explicitly to the entry, because they may be mapped differently.
if (fieldInfos.size() > 1) {
fProcessor->AddFieldToEntry(std::move(valueField), canonicalName, nullptr,
Internal::RNTupleProcessorProvenance());
ROOT::Internal::RNTupleProcessorProvenance());
}

for (std::uint16_t i = 0; i < nFieldInfosAdded; i++) {
fieldInfos.pop_back();
}
}

RNTupleProcessorDS::RNTupleProcessorDS(std::unique_ptr<ROOT::Experimental::RNTupleProcessor> processor)
RNTupleProcessorDS::RNTupleProcessorDS(std::unique_ptr<ROOT::RNTupleProcessor> processor)
: fProcessor(std::move(processor))
{
// Do not add the subfields now, this is handled by AddField.
fProcessor->AddAllFieldsToEntry(Internal::RNTupleProcessorProvenance(), /*includeSubfields=*/false);
fProcessor->AddAllFieldsToEntry(ROOT::Internal::RNTupleProcessorProvenance(), /*includeSubfields=*/false);
const auto &entry = fProcessor->GetEntry();
for (auto fieldIdx : entry.GetFieldIndices()) {
const auto &field = entry.GetValue(fieldIdx).GetField();
Expand Down Expand Up @@ -321,7 +321,7 @@ RNTupleProcessorDS::GetColumnReaders(unsigned int /* slot */, std::string_view n
newField->SetOnDiskId(field.GetOnDiskId());

try {
fProcessor->AddFieldToEntry(std::move(newField), strName, nullptr, Internal::RNTupleProcessorProvenance());
fProcessor->AddFieldToEntry(std::move(newField), strName, nullptr, ROOT::Internal::RNTupleProcessorProvenance());
} catch (const ROOT::RException &) {
std::string msg = "RNTupleProcessorDS: invalid type \"" + requestedType + "\" for column \"" + strName +
"\" with on-disk type \"" + field.GetTypeName() + "\"";
Expand Down Expand Up @@ -375,7 +375,7 @@ bool RNTupleProcessorDS::HasColumn(std::string_view colName) const
void RNTupleProcessorDS::Initialize()
{
fCurrentRanges.clear();
fProcessor->Connect(fProcessor->GetEntry().GetFieldIndices(), Internal::RNTupleProcessorProvenance(),
fProcessor->Connect(fProcessor->GetEntry().GetFieldIndices(), ROOT::Internal::RNTupleProcessorProvenance(),
/*updateFields=*/false);
}

Expand Down Expand Up @@ -425,8 +425,7 @@ ROOT::RDF::RSampleInfo RNTupleProcessorDS::CreateSampleInfo(
}
} // namespace ROOT::Experimental::Internal::RDF

ROOT::RDataFrame
ROOT::Experimental::RDF::FromRNTupleProcessor(std::unique_ptr<ROOT::Experimental::RNTupleProcessor> processor)
ROOT::RDataFrame ROOT::Experimental::RDF::FromRNTupleProcessor(std::unique_ptr<ROOT::RNTupleProcessor> processor)
{
return ROOT::RDataFrame(
std::make_unique<ROOT::Experimental::Internal::RDF::RNTupleProcessorDS>(std::move(processor)));
Expand Down
4 changes: 2 additions & 2 deletions tree/dataframe/test/datasource_ntuple_processor.cxx
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@
using ROOT::TestSupport::FileRaii;

using ROOT::RNTupleModel;
using ROOT::RNTupleProcessor;
using ROOT::RNTupleWriter;
using ROOT::Experimental::CreateRNTupleJoinProcessor;
using ROOT::Experimental::RNTupleProcessor;
using ROOT::Experimental::Internal::RDF::RNTupleProcessorDS;

/*
Expand Down Expand Up @@ -233,7 +233,7 @@ static void ChainTest(const std::string &name, const std::string &fname)
// Causes "Too many open files" errors on Windows
// TODO(fdegeus) Investigate root cause + fix
#if !defined(_MSC_VER) || defined(R__ENABLE_BROKEN_WIN_TESTS)
std::vector<ROOT::Experimental::RNTupleOpenSpec> openSpecs(1000, {name, fname});
std::vector<ROOT::RNTupleOpenSpec> openSpecs(1000, {name, fname});
auto proc1000 = RNTupleProcessor::CreateChain(openSpecs);
auto df1000 = ROOT::Experimental::RDF::FromRNTupleProcessor(std::move(proc1000));
EXPECT_DOUBLE_EQ(42000.0, df1000.Sum<float>("pt").GetValue());
Expand Down
Loading
Loading