23 :
RPageSink(inner->GetNTupleName(), inner->GetWriteOptions())
24 , fMetrics(
"RPageSinkBuf")
25 , fInnerSink(std::move(inner))
29 "compressing pages in parallel")
36 fBufferedColumns.resize(fLastColumnId);
37 fInnerModel = model.
Clone();
38 fInnerSink->Create(*fInnerModel);
54 fBufferedColumns.at(columnHandle.
fId).BufferPage(columnHandle, bufPage);
55 if (!fTaskScheduler) {
58 fCounters->fParallelZip.SetValue(1);
61 zipItem->AllocateSealedPageBuf();
63 fTaskScheduler->AddTask([
this, zipItem, colId = columnHandle.
fId] {
64 zipItem->fSealedPage = SealPage(zipItem->fPage,
65 *fBufferedColumns.at(colId).GetHandle().fColumn->GetElement(),
66 GetWriteOptions().GetCompression(), zipItem->fBuf.get()
79 fInnerSink->CommitSealedPage(columnId, sealedPage);
89 fTaskScheduler->Wait();
90 fTaskScheduler->Reset();
93 for (
auto &bufColumn : fBufferedColumns) {
94 for (
auto &bufPage : bufColumn.DrainBufferedPages()) {
95 if (bufPage.IsSealed()) {
96 fInnerSink->CommitSealedPage(bufColumn.GetHandle().fId, bufPage.fSealedPage);
98 fInnerSink->CommitPage(bufColumn.GetHandle(), bufPage.fPage);
100 ReleasePage(bufPage.fPage);
103 return fInnerSink->CommitCluster(nEntries);
108 fInnerSink->CommitDataset();
114 return fInnerSink->ReservePage(columnHandle, nElements);
119 fInnerSink->ReleasePage(page);
void ObserveMetrics(RNTupleMetrics &observee)
CounterPtrT MakeCounter(const std::string &name, Args &&... args)
A non thread-safe integral performance counter.
std::deque< RPageZipItem >::iterator iterator
RNTupleLocator CommitSealedPageImpl(DescriptorId_t columnId, const RSealedPage &sealedPage) final
RPageSinkBuf(std::unique_ptr< RPageSink > inner)
void ReleasePage(RPage &page) final
Every page store needs to be able to free pages it handed out.
std::unique_ptr< RCounters > fCounters
RPage ReservePage(ColumnHandle_t columnHandle, std::size_t nElements) final
Get a new, empty page for the given column that can be filled with up to nElements.
std::unique_ptr< RPageSink > fInnerSink
The inner sink, responsible for actually performing I/O.
void CreateImpl(const RNTupleModel &model) final
std::uint64_t CommitClusterImpl(NTupleSize_t nEntries) final
Returns the number of bytes written to storage (excluding metadata)
void CommitDatasetImpl() final
RNTupleLocator CommitPageImpl(ColumnHandle_t columnHandle, const RPage &page) final
Abstract interface to write data into an ntuple.
A page is a slice of a column that is mapped into memory.
ClusterSize_t::ValueType GetNElements() const
ClusterSize_t::ValueType GetNBytes() const
The space taken by column elements in the buffer.
void * GrowUnchecked(ClusterSize_t::ValueType nElements)
Called during writing: returns a pointer after the last element and increases the element counter in ...
The RNTupleModel encapulates the schema of an ntuple.
std::unique_ptr< RNTupleModel > Clone() const
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.
I/O performance counters that get registered in fMetrics.
A sealed page contains the bytes of a page as written to storage (packed & compressed).
Generic information about the physical location of data.