Logo ROOT  
Reference Guide
 
Loading...
Searching...
No Matches
RPageStorage.hxx
Go to the documentation of this file.
1/// \file ROOT/RPageStorage.hxx
2/// \author Jakob Blomer <jblomer@cern.ch>
3/// \date 2018-07-19
4
5/*************************************************************************
6 * Copyright (C) 1995-2019, Rene Brun and Fons Rademakers. *
7 * All rights reserved. *
8 * *
9 * For the licensing terms see $ROOTSYS/LICENSE. *
10 * For the list of contributors see $ROOTSYS/README/CREDITS. *
11 *************************************************************************/
12
13#ifndef ROOT_RPageStorage
14#define ROOT_RPageStorage
15
16#include <ROOT/RError.hxx>
17#include <ROOT/RCluster.hxx>
18#include <ROOT/RClusterPool.hxx>
25#include <ROOT/RNTupleTypes.hxx>
26#include <ROOT/RPage.hxx>
27#include <ROOT/RPagePool.hxx>
28#include <ROOT/RSpan.hxx>
29#include <string_view>
30
31#include <atomic>
32#include <cassert>
33#include <cstddef>
34#include <deque>
35#include <functional>
36#include <map>
37#include <memory>
38#include <mutex>
39#include <set>
40#include <shared_mutex>
41#include <unordered_map>
42#include <unordered_set>
43#include <variant>
44#include <vector>
45
46namespace ROOT {
47
48class RNTupleModel;
49
50namespace Internal {
51
52class RPageAllocator;
53class RColumn;
54struct RNTupleModelChangeset;
55
56enum class EPageStorageType {
57 kSink,
58 kSource,
59};
60
61// clang-format off
62/**
63\class ROOT::Internal::RPageStorage
64\ingroup NTuple
65\brief Common functionality of an ntuple storage for both reading and writing
66
67The RPageStore provides access to a storage container that keeps the bits of pages and clusters comprising
68an ntuple. Concrete implementations can use a TFile, a raw file, an object store, and so on.
69*/
70// clang-format on
72public:
73 /// The page checksum is a 64bit xxhash3
74 static constexpr std::size_t kNBytesPageChecksum = sizeof(std::uint64_t);
75
76 /// The interface of a task scheduler to schedule page (de)compression tasks
78 public:
79 virtual ~RTaskScheduler() = default;
80 /// Take a callable that represents a task
81 virtual void AddTask(const std::function<void(void)> &taskFunc) = 0;
82 /// Blocks until all scheduled tasks finished
83 virtual void Wait() = 0;
84 };
85
86 /// A sealed page contains the bytes of a page as written to storage (packed & compressed). It is used
87 /// as an input to UnsealPages() as well as to transfer pages between different storage media.
88 /// RSealedPage does _not_ own the buffer it is pointing to in order to not interfere with the memory management
89 /// of concrete page sink and page source implementations.
90 struct RSealedPage {
91 private:
92 const void *fBuffer = nullptr;
93 std::size_t fBufferSize = 0; ///< Size of the page payload and the trailing checksum (if available)
94 std::uint32_t fNElements = 0;
95 bool fHasChecksum = false; ///< If set, the last 8 bytes of the buffer are the xxhash of the rest of the buffer
96
97 public:
98 RSealedPage() = default;
99 RSealedPage(const void *buffer, std::size_t bufferSize, std::uint32_t nElements, bool hasChecksum = false)
100 : fBuffer(buffer), fBufferSize(bufferSize), fNElements(nElements), fHasChecksum(hasChecksum)
101 {
102 }
103 RSealedPage(const RSealedPage &other) = default;
104 RSealedPage &operator=(const RSealedPage &other) = default;
105 RSealedPage(RSealedPage &&other) = default;
106 RSealedPage &operator=(RSealedPage &&other) = default;
107
108 const void *GetBuffer() const { return fBuffer; }
109 void SetBuffer(const void *buffer) { fBuffer = buffer; }
110
111 std::size_t GetDataSize() const
112 {
115 }
116 std::size_t GetBufferSize() const { return fBufferSize; }
117 void SetBufferSize(std::size_t bufferSize) { fBufferSize = bufferSize; }
118
119 std::uint32_t GetNElements() const { return fNElements; }
120 void SetNElements(std::uint32_t nElements) { fNElements = nElements; }
121
122 bool GetHasChecksum() const { return fHasChecksum; }
123 void SetHasChecksum(bool hasChecksum) { fHasChecksum = hasChecksum; }
124
125 void ChecksumIfEnabled();
127 /// Returns a failure if the sealed page has no checksum
129 };
130
131 using SealedPageSequence_t = std::deque<RSealedPage>;
132 /// A range of sealed pages referring to the same column that can be used for vector commit
135 SealedPageSequence_t::const_iterator fFirst;
136 SealedPageSequence_t::const_iterator fLast;
137
138 RSealedPageGroup() = default;
139 RSealedPageGroup(ROOT::DescriptorId_t d, const SealedPageSequence_t::const_iterator &b,
140 const SealedPageSequence_t::const_iterator &e)
142 {
143 }
144 };
145
146protected:
148
149 /// For the time being, we will use the heap allocator for all sources and sinks. This may change in the future.
150 std::unique_ptr<ROOT::Internal::RPageAllocator> fPageAllocator;
151
152 std::string fNTupleName;
155 {
156 if (!fTaskScheduler)
157 return;
159 }
160
161public:
162 explicit RPageStorage(std::string_view name);
163 RPageStorage(const RPageStorage &other) = delete;
164 RPageStorage &operator=(const RPageStorage &other) = delete;
165 RPageStorage(RPageStorage &&other) = default;
167 virtual ~RPageStorage();
168
169 /// Whether the concrete implementation is a sink or a source
171
175
176 /// Returns true for a valid column handle; fColumn and fPhysicalId should always either both
177 /// be valid or both be invalid.
178 explicit operator bool() const { return fPhysicalId != ROOT::kInvalidDescriptorId && fColumn; }
179 };
180 /// The column handle identifies a column with the current open page storage
182
183 /// Register a new column. When reading, the column must exist in the ntuple on disk corresponding to the metadata.
184 /// When writing, every column can only be attached once.
186 /// Unregisters a column. A page source decreases the reference counter for the corresponding active column.
187 /// For a page sink, dropping columns is currently a no-op.
188 virtual void DropColumn(ColumnHandle_t columnHandle) = 0;
189 ROOT::DescriptorId_t GetColumnId(ColumnHandle_t columnHandle) const { return columnHandle.fPhysicalId; }
190
191 /// Returns the default metrics object. Subclasses might alternatively provide their own metrics object by
192 /// overriding this.
194
195 /// Returns the NTuple name.
196 const std::string &GetNTupleName() const { return fNTupleName; }
197
198 void SetTaskScheduler(RTaskScheduler *taskScheduler) { fTaskScheduler = taskScheduler; }
199}; // class RPageStorage
200
201// clang-format off
202/**
203\class ROOT::Internal::RWritePageMemoryManager
204\ingroup NTuple
205\brief Helper to maintain a memory budget for the write pages of a set of columns
206
207The memory manager keeps track of the sum of bytes used by the write pages of a set of columns.
208It will flush (and shrink) large pages of other columns on the attempt to expand a page.
209*/
210// clang-format on
212private:
213 struct RColumnInfo {
215 std::size_t fCurrentPageSize = 0;
216 std::size_t fInitialPageSize = 0;
217
218 bool operator>(const RColumnInfo &other) const;
219 };
220
221 /// Sum of all the write page sizes (their capacity) of the columns in `fColumnsSortedByPageSize`
222 std::size_t fCurrentAllocatedBytes = 0;
223 /// Maximum allowed value for `fCurrentAllocatedBytes`, set from RNTupleWriteOptions::fPageBufferBudget
224 std::size_t fMaxAllocatedBytes = 0;
225 /// All columns that called `ReservePage()` (hence `TryUpdate()`) at least once,
226 /// sorted by their current write page size from large to small
227 std::set<RColumnInfo, std::greater<RColumnInfo>> fColumnsSortedByPageSize;
228
229 /// Flush columns in order of allocated write page size until the sum of all write page allocations
230 /// leaves space for at least targetAvailableSize bytes. Only use columns with a write page size larger
231 /// than pageSizeLimit.
232 bool TryEvict(std::size_t targetAvailableSize, std::size_t pageSizeLimit);
233
234public:
235 explicit RWritePageMemoryManager(std::size_t maxAllocatedBytes) : fMaxAllocatedBytes(maxAllocatedBytes) {}
236
237 /// Try to register the new write page size for the given column. Flush large columns to make space, if necessary.
238 /// If not enough space is available after all (sum of write pages would be larger than fMaxAllocatedBytes),
239 /// return false.
240 bool TryUpdate(ROOT::Internal::RColumn &column, std::size_t newWritePageSize);
241};
242
243// clang-format off
244/**
245\class ROOT::Internal::RPageSink
246\ingroup NTuple
247\brief Abstract interface to write data into an ntuple
248
249The page sink takes the list of columns and afterwards a series of page commits and cluster commits.
250The user is responsible to commit clusters at a consistent point, i.e. when all pages corresponding to data
251up to the given entry number are committed.
252
253An object of this class may either be a wrapper (for example a RPageSinkBuf) or a "persistent" sink,
254inheriting from RPagePersistentSink.
255*/
256// clang-format on
257class RPageSink : public RPageStorage {
258public:
259 using Callback_t = std::function<void(RPageSink &)>;
260
261 /// Cluster that was staged, but not yet logically appended to the RNTuple
275
276protected:
277 std::unique_ptr<ROOT::RNTupleWriteOptions> fOptions;
278
279 /// Flag if sink was initialized
280 bool fIsInitialized = false;
281
282 /// Helper for streaming a page. This is commonly used in derived, concrete page sinks. Note that if
283 /// compressionSetting is 0 (uncompressed) and the page is mappable and not checksummed, the returned sealed page
284 /// will point directly to the input page buffer. Otherwise, the sealed page references fSealPageBuffer. Thus,
285 /// the buffer pointed to by the RSealedPage should never be freed.
287
288private:
289 std::vector<Callback_t> fOnDatasetCommitCallbacks;
290 std::vector<unsigned char> fSealPageBuffer; ///< Used as destination buffer in the simple SealPage overload
291
292 /// Used in ReservePage to maintain the page buffer budget
294
295public:
296 RPageSink(std::string_view ntupleName, const ROOT::RNTupleWriteOptions &options);
297
298 RPageSink(const RPageSink &) = delete;
299 RPageSink &operator=(const RPageSink &) = delete;
300 RPageSink(RPageSink &&) = default;
302 ~RPageSink() override;
303
305 /// Returns the sink's write options.
307
308 void DropColumn(ColumnHandle_t /*columnHandle*/) final {}
309
310 bool IsInitialized() const { return fIsInitialized; }
311
312 /// Return the RNTupleDescriptor being constructed.
313 virtual const ROOT::RNTupleDescriptor &GetDescriptor() const = 0;
314
315 virtual ROOT::NTupleSize_t GetNEntries() const = 0;
316
317 /// Physically creates the storage container to hold the ntuple (e.g., a keys a TFile or an S3 bucket)
318 /// Init() associates column handles to the columns referenced by the model
319 void Init(RNTupleModel &model)
320 {
321 if (fIsInitialized) {
322 throw RException(R__FAIL("already initialized"));
323 }
324 InitImpl(model);
325 fIsInitialized = true;
326 }
327
328protected:
329 virtual void InitImpl(RNTupleModel &model) = 0;
331
332public:
333 /// Parameters for the SealPage() method
335 const ROOT::Internal::RPage *fPage = nullptr; ///< Input page to be sealed
337 nullptr; ///< Corresponds to the page's elements, for size calculation etc.
338 std::uint32_t fCompressionSettings = 0; ///< Compression algorithm and level to apply
339 /// Adds a 8 byte little-endian xxhash3 checksum to the page payload. The buffer has to be large enough to
340 /// to store the additional 8 bytes.
341 bool fWriteChecksum = true;
342 /// If false, the output buffer must not point to the input page buffer, which would otherwise be an option
343 /// if the page is mappable and should not be compressed
344 bool fAllowAlias = false;
345 /// Location for sealed output. The memory buffer has to be large enough.
346 void *fBuffer = nullptr;
347 };
348
349 /// Seal a page using the provided info.
350 static RSealedPage SealPage(const RSealPageConfig &config);
351
352 /// Incorporate incremental changes to the model into the ntuple descriptor. This happens, e.g. if new fields were
353 /// added after the initial call to `RPageSink::Init(RNTupleModel &)`.
354 /// `firstEntry` specifies the global index for the first stored element in the added columns.
355 virtual void UpdateSchema(const ROOT::Internal::RNTupleModelChangeset &changeset, ROOT::NTupleSize_t firstEntry) = 0;
356 /// Adds an extra type information record to schema. The extra type information will be written to the
357 /// extension header. The information in the record will be merged with the existing information, e.g.
358 /// duplicate streamer info records will be removed. This method is called by the "on commit dataset" callback
359 /// registered by specific fields (e.g., streamer field) and during merging.
360 virtual void UpdateExtraTypeInfo(const ROOT::RExtraTypeInfoDescriptor &extraTypeInfo) = 0;
361
362 /// Commits a suppressed column for the current cluster. Can be called anytime before CommitCluster().
363 /// For any given column and cluster, there must be no calls to both CommitSuppressedColumn() and page commits.
364 virtual void CommitSuppressedColumn(ColumnHandle_t columnHandle) = 0;
365 /// Write a page to the storage. The column must have been added before.
366 virtual void CommitPage(ColumnHandle_t columnHandle, const ROOT::Internal::RPage &page) = 0;
367 /// Write a preprocessed page to storage. The column must have been added before.
368 virtual void
369 CommitSealedPage(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage) = 0;
370 /// Write a vector of preprocessed pages to storage. The corresponding columns must have been added before.
371 virtual void CommitSealedPageV(std::span<RPageStorage::RSealedPageGroup> ranges) = 0;
372 /// Stage the current cluster and create a new one for the following data.
373 /// Returns the object that must be passed to CommitStagedClusters to logically append the staged cluster to the
374 /// ntuple descriptor.
376 /// Commit staged clusters, logically appending them to the ntuple descriptor.
377 virtual void CommitStagedClusters(std::span<RStagedCluster> clusters) = 0;
378 /// Finalize the current cluster and create a new one for the following data.
379 /// Returns the number of bytes written to storage (excluding metadata).
380 virtual std::uint64_t CommitCluster(ROOT::NTupleSize_t nNewEntries)
381 {
382 RStagedCluster stagedClusters[] = {StageCluster(nNewEntries)};
383 CommitStagedClusters(stagedClusters);
384 return stagedClusters[0].fNBytesWritten;
385 }
386 /// Write out the page locations (page list envelope) for all the committed clusters since the last call of
387 /// CommitClusterGroup (or the beginning of writing).
388 virtual void CommitClusterGroup() = 0;
389
390 /// Creates a new sink with the same underlying storage as this but writing to a different
391 /// RNTuple named `name`. Only one of the two sinks can safely write at the same time.
392 /// The RNTuple written by this cloned sink will be stored in a hidden key (this is a convenient assumption we
393 /// make now since this method is only used to create attribute RNTuples).
394 virtual std::unique_ptr<RPageSink> CloneAsHidden(std::string_view name, const RNTupleWriteOptions &opts) const = 0;
395
396 /// Adds the given anchor information (name + locator) into the main RNTuple's descriptor as an attribute set
397 /// linked to it with the given name.
398 /// The attribute set must have already been written to storage via RNTupleAttrSetWriter::Commit().
399 /// Note that, by RNTuple specs, this is only legal to call on a non-attribute RNTuple's sink.
400 virtual void CommitAttributeSet(std::string_view attrSetName, const RNTupleLink &attrAnchorInfo) = 0;
401
402 /// The registered callback is executed at the beginning of CommitDataset();
404 /// Run the registered callbacks and finalize the current cluster and the entrire data set.
406
407 /// Get a new, empty page for the given column that can be filled with up to nElements;
408 /// nElements must be larger than zero.
409 virtual ROOT::Internal::RPage ReservePage(ColumnHandle_t columnHandle, std::size_t nElements);
410
411 /// An RAII wrapper used to synchronize a page sink. See GetSinkGuard().
413 std::mutex *fLock;
414
415 public:
416 explicit RSinkGuard(std::mutex *lock) : fLock(lock)
417 {
418 if (fLock != nullptr) {
419 fLock->lock();
420 }
421 }
422 RSinkGuard(const RSinkGuard &) = delete;
423 RSinkGuard &operator=(const RSinkGuard &) = delete;
424 RSinkGuard(RSinkGuard &&) = delete;
427 {
428 if (fLock != nullptr) {
429 fLock->unlock();
430 }
431 }
432 };
433
435 {
436 // By default, there is no lock and the guard does nothing.
437 return RSinkGuard(nullptr);
438 }
439}; // class RPageSink
440
441// clang-format off
442/**
443\class ROOT::Internal::RPagePersistentSink
444\ingroup NTuple
445\brief Base class for a sink with a physical storage backend
446
447Call RNTupleWriter::CommitDataset() explicitly rather than relying on the writer's destructor. A destructor
448cannot throw, so if committing fails there the error is only logged and the ntuple may be left incomplete
449on storage.
450*/
451// clang-format on
453private:
454 /// Used to map the IDs of the descriptor to the physical IDs issued during header/footer serialization
456
457 /// Remembers the starting cluster id for the next cluster group
458 std::uint64_t fNextClusterInGroup = 0;
459 /// Used to calculate the number of entries in the current cluster
461 /// Keeps track of the number of elements in the currently open cluster. Indexed by column id.
462 std::vector<ROOT::RClusterDescriptor::RColumnRange> fOpenColumnRanges;
463 /// Keeps track of the written pages in the currently open cluster. Indexed by column id.
464 std::vector<ROOT::RClusterDescriptor::RPageRange> fOpenPageRanges;
465
466 /// Union of the streamer info records that are sent from streamer fields to the sink before committing the dataset.
468
469protected:
470 /// Set of optional features supported by the persistent sink
471 struct RFeatures {
472 bool fCanMergePages = false;
473 };
474
477
478 /// Default I/O performance counters that get registered in fMetrics
488 std::unique_ptr<RCounters> fCounters;
489
490 virtual void InitImpl(unsigned char *serializedHeader, std::uint32_t length) = 0;
491
492 /// Updates the descriptor and calls InitImpl() that handles the backend-specific details (file, DAOS, etc.)
493 void InitImpl(RNTupleModel &model) final;
494
495 virtual RNTupleLocator
497 /// Vector commit of preprocessed pages. The `ranges` array specifies a range of sealed pages to be
498 /// committed for each column. The returned vector contains, in order, the RNTupleLocator for each
499 /// page on each range in `ranges`, i.e. the first N entries refer to the N pages in `ranges[0]`,
500 /// followed by M entries that refer to the M pages in `ranges[1]`, etc.
501 /// The mask allows to skip writing out certain pages. The vector has the size of all the pages.
502 /// For every `false` value in the mask, the corresponding locator is skipped (missing) in the output vector.
503 /// The default is to call `CommitSealedPageImpl` for each page; derived classes may provide an
504 /// optimized implementation though.
505 virtual std::vector<RNTupleLocator>
506 CommitSealedPageVImpl(std::span<RPageStorage::RSealedPageGroup> ranges, const std::vector<bool> &mask);
507 /// Returns the number of bytes written to storage (excluding metadata)
508 virtual std::uint64_t StageClusterImpl() = 0;
509 /// Returns the locator of the page list envelope of the given buffer that contains the serialized page list.
510 /// Typically, the implementation takes care of compressing and writing the provided buffer.
511 virtual RNTupleLocator CommitClusterGroupImpl(unsigned char *serializedPageList, std::uint32_t length) = 0;
512 virtual RNTupleLink CommitDatasetImpl(unsigned char *serializedFooter, std::uint32_t length) = 0;
513
514 /// \return The locator and length of the written anchor.
516
517 /// Enables the default set of metrics provided by RPageSink. `prefix` will be used as the prefix for
518 /// the counters registered in the internal RNTupleMetrics object.
519 /// This set of counters can be extended by a subclass by calling `fMetrics.MakeCounter<...>()`.
520 ///
521 /// A subclass using the default set of metrics is always responsible for updating the counters
522 /// appropriately, e.g. `fCounters->fNPageCommited.Inc()`
523 void EnableDefaultMetrics(const std::string &prefix);
524
525public:
526 RPagePersistentSink(std::string_view ntupleName, const ROOT::RNTupleWriteOptions &options);
527
529 RPagePersistentSink &operator=(const RPagePersistentSink &) = delete;
531 RPagePersistentSink &operator=(RPagePersistentSink &&) = default;
532 ~RPagePersistentSink() override;
533
534 /// Guess the concrete derived page source from the location
535 static std::unique_ptr<RPageSink> Create(std::string_view ntupleName, std::string_view location,
536 const ROOT::RNTupleWriteOptions &options = ROOT::RNTupleWriteOptions());
537
538 ColumnHandle_t AddColumn(ROOT::DescriptorId_t fieldId, ROOT::Internal::RColumn &column) final;
539
541
543
544 void UpdateSchema(const ROOT::Internal::RNTupleModelChangeset &changeset, ROOT::NTupleSize_t firstEntry) override;
545 void UpdateExtraTypeInfo(const ROOT::RExtraTypeInfoDescriptor &extraTypeInfo) final;
546
547 /// Initialize sink based on an existing descriptor and fill into the descriptor builder, optionally copying over
548 /// the descriptor's clusters to this sink's descriptor.
549 /// \return The model created from the new sink's descriptor. This model should be kept alive
550 /// for at least as long as the sink.
551 [[nodiscard]] std::unique_ptr<RNTupleModel>
552 InitFromDescriptor(const ROOT::RNTupleDescriptor &descriptor, bool copyClusters);
553
554 /// Adds a new column representation to the given field. It is NOT marked as suppressed.
555 /// The first element index of the added column will be computed as the sum of the first element index of the
556 /// 0th representation's column range plus `addedFirstElementIndex` (which must be non-negative).
557 /// This can only be legally done at cluster boundary (i.e. immediately after committing a cluster or before
558 /// writing any entry). This is NOT checked for you.
559 /// `newRepresentation` must have a number of elements equal to the field's cardinality and must contain
560 /// compatible column types.
561 /// NOTE: this function was designed for use by the Merger and it's not fit as-is for use with the higher level
562 /// API layers. In other words, you cannot use this with a Writer and write data through the regular Entry/Field
563 /// API, as this function only modifies the descriptor.
564 /// \return The physical id of the first newly added column.
566 std::span<const ROOT::Internal::RColumnFormat> newRepresentation,
567 std::uint64_t clusterOffset);
568
569 /// Adds a new alias column pointing to an existing column with the given physical id to the given field.
571 ROOT::DescriptorId_t physicalId);
572
573 void CommitSuppressedColumn(ColumnHandle_t columnHandle) final;
574 void CommitPage(ColumnHandle_t columnHandle, const ROOT::Internal::RPage &page) final;
575 void CommitSealedPage(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage) final;
576 void CommitSealedPageV(std::span<RPageStorage::RSealedPageGroup> ranges) final;
577 RStagedCluster StageCluster(ROOT::NTupleSize_t nNewEntries) final;
578 void CommitStagedClusters(std::span<RStagedCluster> clusters) final;
579 void CommitClusterGroup() final;
580 void CommitAttributeSet(std::string_view attrSetName, const RNTupleLink &attrAnchorInfo) final;
581}; // class RPagePersistentSink
582
583// clang-format off
584/**
585\class ROOT::Internal::RPageSource
586\ingroup NTuple
587\brief Abstract interface to read data from an ntuple
588
589The page source is initialized with the columns of interest. Alias columns from projected fields are mapped to the
590corresponding physical columns. Pages from the columns of interest can then be mapped into memory.
591The page source also gives access to the ntuple's metadata.
592*/
593// clang-format on
594class RPageSource : public RPageStorage {
595 // Decompress pages smaller than 4kB directly in the main thread, not in a TBB thread
596 static constexpr std::size_t kInlineDecompressionThreshold = 4096;
597
598 /// Summarizes meta-data necessary to load a certain page. Used by LoadPageFromSummary().
600 ROOT::DescriptorId_t fClusterId = 0;
601 /// The first element number of the page's column in the given cluster
602 std::uint64_t fColumnOffset = 0;
603 /// Location of the page on disk
605 };
606
607public:
608 /// Used in SetEntryRange / GetEntryRange
609 struct REntryRange {
611 ROOT::NTupleSize_t fNEntries = 0;
612
613 /// Returns true if the given cluster has entries within the entry range
614 bool IntersectsWith(const ROOT::RClusterDescriptor &clusterDesc) const;
615 };
616
617 /// An RAII wrapper used for the read-only access to `RPageSource::fDescriptor`. See `GetExclDescriptorGuard()``.
619 const ROOT::RNTupleDescriptor *fDescriptor = nullptr;
620 std::shared_mutex *fLock = nullptr;
621
622 public:
624 RSharedDescriptorGuard(const ROOT::RNTupleDescriptor &desc, std::shared_mutex &lock)
625 : fDescriptor(&desc), fLock(&lock)
626 {
627 fLock->lock_shared();
628 }
632 {
633 std::swap(fDescriptor, other.fDescriptor);
634 std::swap(fLock, other.fLock);
635 }
637 {
638 if (this == &other)
639 return *this;
640 std::swap(fDescriptor, other.fDescriptor);
641 std::swap(fLock, other.fLock);
642 return *this;
643 }
645 {
646 if (IsValid())
647 Release();
648 }
649 const ROOT::RNTupleDescriptor *operator->() const { return fDescriptor; }
650 const ROOT::RNTupleDescriptor &GetConstRef() const { return *fDescriptor; }
651 bool IsValid() const { return fLock; }
652 void Release()
653 {
654 fLock->unlock_shared();
655 fLock = nullptr;
656 fDescriptor = nullptr;
657 }
658 };
659
660 /// An RAII wrapper used for the writable access to `RPageSource::fDescriptor`. See `GetSharedDescriptorGuard()`.
662 ROOT::RNTupleDescriptor *fDescriptor = nullptr;
663 std::shared_mutex *fLock = nullptr;
664
665 public:
667 RExclDescriptorGuard(ROOT::RNTupleDescriptor &desc, std::shared_mutex &lock) : fDescriptor(&desc), fLock(&lock)
668 {
669 fLock->lock();
670 }
674 {
675 std::swap(fDescriptor, other.fDescriptor);
676 std::swap(fLock, other.fLock);
677 }
679 {
680 if (this == &other)
681 return *this;
682 std::swap(fDescriptor, other.fDescriptor);
683 std::swap(fLock, other.fLock);
684 return *this;
685 }
687 {
688 if (IsValid())
689 Release();
690 }
691 ROOT::RNTupleDescriptor &operator*() const { return *fDescriptor; }
692 ROOT::RNTupleDescriptor *operator->() const { return fDescriptor; }
693 void MoveIn(ROOT::RNTupleDescriptor desc) { *fDescriptor = std::move(desc); }
694 bool IsValid() const { return fLock; }
695 void Release()
696 {
697 fDescriptor->IncGeneration();
698 fLock->unlock();
699 fLock = nullptr;
700 fDescriptor = nullptr;
701 }
702 };
703
704 /// Either a shared or an exclusive lock guard
706 private:
707 std::variant<RSharedDescriptorGuard, RExclDescriptorGuard> fAnyGuard;
708
709 public:
710 RAnyDescriptorGuard(RSharedDescriptorGuard sharedGuard) : fAnyGuard(std::move(sharedGuard)) {}
711 RAnyDescriptorGuard(RExclDescriptorGuard exclGuard) : fAnyGuard(std::move(exclGuard)) {}
717
719 {
720 return (fAnyGuard.index() == 0) ? std::get<0>(fAnyGuard).GetConstRef() : std::get<1>(fAnyGuard).operator*();
721 }
723 {
724 return (fAnyGuard.index() == 0) ? std::get<0>(fAnyGuard).operator->() : std::get<1>(fAnyGuard).operator->();
725 }
726 bool IsValid() const
727 {
728 return (fAnyGuard.index() == 0) ? std::get<0>(fAnyGuard).IsValid() : std::get<1>(fAnyGuard).IsValid();
729 }
730 void Release()
731 {
732 if (fAnyGuard.index() == 0)
733 std::get<0>(fAnyGuard).Release();
734 else
735 std::get<1>(fAnyGuard).Release();
736 }
737 };
738
739private:
740 /// Keeps track of the requested physical column IDs and their in-memory target type via a column element identifier.
741 /// When using alias columns (projected fields), physical columns may be requested multiple times.
743 public:
748
749 private:
750 /// Maps physical column IDs to all the requested in-memory representations.
751 /// A pair of physical column ID and in-memory representation can be requested multiple times, which is
752 /// indicated by the reference counter.
753 /// We can only have a handful of possible in-memory representations for a given column,
754 /// so it is fine to search them linearly.
755 std::unordered_map<ROOT::DescriptorId_t, std::vector<RColumnInfo>> fColumnInfos;
756
757 public:
758 void Insert(ROOT::DescriptorId_t physicalColumnId, ROOT::Internal::RColumnElementBase::RIdentifier elementId);
759 void Erase(ROOT::DescriptorId_t physicalColumnId, ROOT::Internal::RColumnElementBase::RIdentifier elementId);
760 ROOT::Internal::RCluster::ColumnSet_t ToColumnSet() const;
761 bool HasColumnInfos(ROOT::DescriptorId_t physicalColumnId) const
762 {
763 return fColumnInfos.count(physicalColumnId) > 0;
764 }
765 const std::vector<RColumnInfo> &GetColumnInfos(ROOT::DescriptorId_t physicalColumnId) const
766 {
767 return fColumnInfos.at(physicalColumnId);
768 }
769 };
770
772 mutable std::shared_mutex fDescriptorLock;
773 REntryRange fEntryRange; ///< Used by the cluster pool to prevent reading beyond the given range
774 bool fHasStructure = false; ///< Set to true once `LoadStructure()` is called
775 bool fIsAttached = false; ///< Set to true once `Attach()` is called
776 bool fHasStreamerInfosRegistered = false; ///< Set to true when RegisterStreamerInfos() is called.
777
778 /// The interpretation of page lists, set in Attach()
781
782 /// This vector is aligned with the cluster groups in the descriptor. It stores the cumulative number of clusters
783 /// in the cluster groups. Given a cluster ID, we can thus quickly determine the cluster group that the cluster
784 /// comes from. While descriptor IDs in a descriptor in general are arbitrary, for the descriptor in the page source,
785 /// that was created from a serialized on-disk representation, we know that cluster and cluster group IDs are
786 /// issued consecutively.
787 std::vector<NTupleSize_t> fCumulativeClusterCounts;
788
789 /// The active columns are implicitly defined by the model fields or views
791
792 /// The cluster pool asynchronously preloads the next few clusters. Note that derived classes should call
793 /// StopClusterPoolBackgroundThread() in their destructor so that the I/O background thread does not call
794 /// methods from the destructed derived class.
796 /// The last cluster from which a page got loaded. Points into fClusterPool->fPool
797 ROOT::Internal::RCluster *fCurrentCluster = nullptr;
798 /// Pages that are unzipped with IMT are staged into the page pool
800
801 /// Remembers the last cluster id from which a page was requested
803 /// Clusters from where pages got preloaded in UnzipCluster(), ordered by first entry number
804 /// of the clusters. If the last used cluster changes in LoadPage(), all unused pages from
805 /// previous clusters are evicted from the page pool. Pinned clusters won't be evicted.
806 std::map<ROOT::NTupleSize_t, ROOT::DescriptorId_t> fPreloadedClusters;
807
808 /// Pinned clusters and their $2 * (cluster bunch size) - 1$ successors will not be evicted from the cluster pool.
809 /// Pages of pinned clusters won't be evicted from the page pool.
810 std::unordered_set<ROOT::DescriptorId_t> fPinnedClusters;
811
812 /// Ensures that fDescriptor has cluster details loaded for the given cluster group ID. This method is expected
813 /// to be called while holding the lock passed by descGuard. The returned descriptor guard is either the
814 /// passed one or a new, exclusive guard if cluster details needed to be loaded.
815 RAnyDescriptorGuard EnsureClusterDetails(DescriptorId_t cgId, RAnyDescriptorGuard descGuard);
816 /// Uses binary search in fCumulativeClusterCounts to determine the cluster group that the cluster ID belongs in.
817 DescriptorId_t FindClusterGroupId(DescriptorId_t clusterId) const;
818
819 /// Does nothing if fLastUsedCluster == clusterId. Otherwise, updated fLastUsedCluster
820 /// and evict unused paged from the page pool of all previous clusters.
821 /// Must not be called when the descriptor guard is taken.
822 void UpdateLastUsedCluster(ROOT::DescriptorId_t clusterId);
823
824 // Populate the cluster details of the given cluster group. Must hold an exclusive descriptor lock when calling.
825 // No-op if the cluster details are already present.
826 void LoadPageList(DescriptorId_t clusterGroupId, const RExclDescriptorGuard &exclGuard);
827 // Common treatment of zero pages in LoadPageFromSummary()
828 ROOT::Internal::RPageRef LoadZeroPage(ColumnHandle_t columnHandle, const RPageSummary &pageSummary);
829 // Once the page is found to be missing in the page cache and all information about the page is collected,
830 // either using a global or a local element index, perform the common page loading tasks using the page summary.
831 // This includes zero page handling and prefetching the corresponding cluster.
832 ROOT::Internal::RPageRef LoadPageFromSummary(ColumnHandle_t columnHandle, const RPageSummary &pageSummary);
833
834protected:
835 /// Holds the uncompressed header and footer
837 std::unique_ptr<unsigned char[]> fBuffer; ///< single buffer for both header and footer
838 void *fPtrHeader = nullptr; ///< either nullptr or points into fBuffer
839 void *fPtrFooter = nullptr; ///< either nullptr or points into fBuffer
840
841 /// Called at the end of Attach(), i.e. when the header and footer are processed
842 void Reset()
843 {
844 RStructureBuffer empty;
845 std::swap(empty, *this);
846 }
847 };
848
849 /// Default I/O performance counters that get registered in `fMetrics`
869
870 std::unique_ptr<RCounters> fCounters;
871 RStructureBuffer fStructureBuffer; ///< Populated by LoadStructureImpl(), reset at the end of Attach()
872
874
875 /// Fills fStructureBuffer with the compressed header and footer
876 virtual void LoadStructureImpl() = 0;
877 /// `LoadStructureImpl()` has been called before `AttachImpl()` is called
879 /// Returns a new, unattached page source for the same data set. If the origin data source was attached, the
880 /// base class Clone() method will as an optimization not call LoadStructureImpl() and AttachImpl() again
881 /// but simply clone the existing descriptor. This means that CloneImpl(), for an attached page source, needs to
882 /// redo any operation other than loading the descriptor that may be done in LoadStructureImpl() and AttachImpl().
883 virtual std::unique_ptr<RPageSource> CloneImpl() const = 0;
884 // Loads a page list into the provided buffer. The buffer parameter needs to point to a memory region
885 // that has space for at least locator.GetNBytesOnStorage() bytes to hold the compressed page list.
886 virtual void LoadPageListImpl(const RNTupleLocator &locator, unsigned char *buffer) = 0;
887 // Returns a sealed page from storage without adding it to the page pool. The sealed pages buffer and buffer size
888 // is already initialized.
889 virtual void LoadSealedPageImpl(const RNTupleLocator &locator, RSealedPage &sealedPage) = 0;
890
891 /// Prepare a page range read for the column set in `clusterKey`. Specifically, pages referencing the
892 /// `kTypePageZero` locator are filled in `pageZeroMap`; otherwise, `perPageFunc` is called for each page. This is
893 /// commonly used as part of `LoadClusters()` in derived classes.
894 void PrepareLoadCluster(const ROOT::Internal::RCluster::RKey &clusterKey,
896 const std::function<void(ROOT::DescriptorId_t, ROOT::NTupleSize_t,
897 const ROOT::RClusterDescriptor::RPageInfo &)> &perPageFunc);
898
899 /// Enables the default set of metrics provided by RPageSource. `prefix` will be used as the prefix for
900 /// the counters registered in the internal RNTupleMetrics object.
901 /// A subclass using the default set of metrics is responsible for updating the counters
902 /// appropriately, e.g. `fCounters->fNRead.Inc()`
903 /// Alternatively, a subclass might provide its own RNTupleMetrics object by overriding the
904 /// `GetMetrics()` member function.
905 void EnableDefaultMetrics(const std::string &prefix);
906
907 /// Note that the underlying lock is not recursive. See `GetSharedDescriptorGuard()` for further information.
908 RExclDescriptorGuard GetExclDescriptorGuard() { return RExclDescriptorGuard(fDescriptor, fDescriptorLock); }
909
910 // To be called in the destructor of derived classes
912
913public:
914 RPageSource(std::string_view ntupleName, const ROOT::RNTupleReadOptions &fOptions);
915 RPageSource(const RPageSource &) = delete;
919 ~RPageSource() override;
920 /// Guess the concrete derived page source from the file name (location)
921 static std::unique_ptr<RPageSource> Create(std::string_view ntupleName, std::string_view location,
923 /// Open the same storage multiple time, e.g. for reading in multiple threads.
924 /// If the source is already attached, the clone will be attached, too. The clone will use, however,
925 /// it's own connection to the underlying storage (e.g., file descriptor, XRootD handle, etc.)
926 std::unique_ptr<RPageSource> Clone() const;
927
928 /// Helper for unstreaming a page. This is commonly used in derived, concrete page sources. The implementation
929 /// currently always makes a memory copy, even if the sealed page is uncompressed and in the final memory layout.
930 /// The optimization of directly mapping pages is left to the concrete page source implementations.
931 RResult<ROOT::Internal::RPage> static UnsealPage(const RSealedPage &sealedPage,
934
937
938 /// Takes the read lock for the descriptor. Multiple threads can take the lock concurrently.
939 /// The underlying `std::shared_mutex`, however, is neither read nor write recursive:
940 /// within one thread, only one lock (shared or exclusive) must be acquired at the same time. This requires special
941 /// care in sections protected by `GetSharedDescriptorGuard()` and `GetExclDescriptorGuard()` especially to avoid
942 /// that the locks are acquired indirectly. As a general guideline, no other
943 /// method of the page source should be called (directly or indirectly) in a guarded section.
945 {
946 return RSharedDescriptorGuard(fDescriptor, fDescriptorLock);
947 }
948
950 void DropColumn(ColumnHandle_t columnHandle) override;
951
952 /// Loads header and footer without decompressing or deserializing them. This can be used to asynchronously open
953 /// a file in the background. The method is idempotent and it is called as a first step in `Attach()`.
954 /// Pages sources may or may not make use of splitting loading and processing metadata.
955 /// Therefore, `LoadStructure()` may do nothing and defer loading the metadata to `Attach()`.
956 void LoadStructure();
957 /// Open the physical storage container and deserialize header and footer
960 /// Load all remaining cluster details
961 void LoadAllPageLists();
962
964 ROOT::NTupleSize_t GetNElements(ROOT::DescriptorId_t physicalColumnId);
965 /// Returns a shared descriptor guard to ensure that the returned cluster id is useable, i.e. that the
966 /// corresponding cluster was not meanwhile evicted from the set of active clusters.
967 RAnyDescriptorGuard
968 FindClusterId(ROOT::DescriptorId_t physicalColumnId, ROOT::NTupleSize_t index, ROOT::DescriptorId_t &cid);
969 /// An overload of FindClusterId that searches using a certain column element index.
970 RAnyDescriptorGuard FindClusterId(ROOT::NTupleSize_t entryIdx, ROOT::DescriptorId_t &cid);
971 /// Uses FindClusterId to search for the cluster with the entry index following the last entry index of the
972 /// given cluster.
973 RAnyDescriptorGuard FindNextClusterId(ROOT::DescriptorId_t clusterId, ROOT::DescriptorId_t &nextId);
974
975 /// Promise to only read from the given entry range. If set, prevents the cluster pool from reading-ahead beyond
976 /// the given range. The range needs to be within `[0, GetNEntries())`.
977 void SetEntryRange(const REntryRange &range);
978 REntryRange GetEntryRange() const { return fEntryRange; }
979
980 /// Allocates and fills a page that contains the index-th element. Calls into the concrete page source
981 /// for loading the corresponding sealed page of cluster where necessary.
982 /// Returns a default-constructed RPage for suppressed columns.
983 virtual ROOT::Internal::RPageRef LoadPage(ColumnHandle_t columnHandle, ROOT::NTupleSize_t globalIndex);
984 /// Another version of `LoadPage` that allows to specify cluster-relative indexes.
985 /// Returns a default-constructed RPage for suppressed columns.
986 virtual ROOT::Internal::RPageRef LoadPage(ColumnHandle_t columnHandle, RNTupleLocalIndex localIndex);
987
988 /// Read the packed and compressed bytes of a page into the memory buffer provided by `sealedPage`. The sealed page
989 /// can be used subsequently in a call to `RPageSink::CommitSealedPage`.
990 /// The `fSize` and `fNElements` member of the sealedPage parameters are always set. If `sealedPage.fBuffer` is
991 /// `nullptr`, no data will be copied but the returned size information can be used by the caller to allocate a large
992 /// enough buffer and call `LoadSealedPage` again.
993 void LoadSealedPage(ROOT::DescriptorId_t physicalColumnId, RNTupleLocalIndex localIndex, RSealedPage &sealedPage);
994
995 /// Populates all the pages of the given cluster ids and columns; it is possible that some columns do not
996 /// contain any pages. The page source may load more columns than the minimal necessary set from `columns`.
997 /// To indicate which columns have been loaded, `LoadClusters()`` must mark them with `SetColumnAvailable()`.
998 /// That includes the ones from the `columns` that don't have pages; otherwise subsequent requests
999 /// for the cluster would assume an incomplete cluster and trigger loading again.
1000 /// `LoadClusters()` is typically called from the I/O thread of a cluster pool, i.e. the method runs
1001 /// concurrently to other methods of the page source.
1002 virtual std::vector<std::unique_ptr<ROOT::Internal::RCluster>>
1003 LoadClusters(std::span<ROOT::Internal::RCluster::RKey> clusterKeys) = 0;
1004
1005 /// Parallel decompression and unpacking of the pages in the given cluster. The unzipped pages are supposed
1006 /// to be preloaded in a page pool attached to the source. The method is triggered by the cluster pool's
1007 /// unzip thread. It is an optional optimization, the method can safely do nothing. In particular, the
1008 /// unzipping will only run if a task scheduler is set. In practice, a task scheduler is set if
1009 /// implicit multi-threading is turned on.
1010 void UnzipCluster(ROOT::Internal::RCluster *cluster);
1011
1012 /// Instructs the cluster pool and page pool to consider the given cluster as active (should stay cached).
1013 void PinCluster(ROOT::DescriptorId_t clusterId) { fPinnedClusters.insert(clusterId); }
1014 /// Allows the given cluster to be evicted from the cluster pool and page pool.
1015 void UnpinCluster(ROOT::DescriptorId_t clusterId) { fPinnedClusters.erase(clusterId); }
1016 const std::unordered_set<ROOT::DescriptorId_t> &GetPinnedClusters() const { return fPinnedClusters; }
1017
1018 // TODO(gparolini): for symmetry with SealPage(), we should either make this private or SealPage() public.
1020 UnsealPage(const RSealedPage &sealedPage, const ROOT::Internal::RColumnElementBase &element);
1021
1022 /// Builds the streamer info records from the descriptor's extra type info section. This is necessary when
1023 /// connecting streamer fields so that emulated classes can be read.
1024 void RegisterStreamerInfos();
1025
1026 /// Creates a new PageSource using the same underlying file as this but referring to a different RNTuple,
1027 /// described by `anchorLink`.
1028 virtual std::unique_ptr<RPageSource> OpenWithDifferentAnchor(const ROOT::Internal::RNTupleLink &anchorLink,
1029 const ROOT::RNTupleReadOptions &options = {}) = 0;
1030}; // class RPageSource
1031
1032} // namespace Internal
1033} // namespace ROOT
1034
1035#endif
#define R__FAIL(msg)
Short-hand to return an RResult<T> in an error state; the RError is implicitly converted into RResult...
Definition RError.hxx:322
#define d(i)
Definition RSha256.hxx:102
#define b(i)
Definition RSha256.hxx:100
#define e(i)
Definition RSha256.hxx:103
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 index
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 mode
char name[80]
Definition TGX11.cxx:142
A thread-safe integral performance counter.
A metric element that computes its floating point value from other counters.
A collection of Counter objects with a name, a unit, and a description.
Managed a set of clusters containing compressed and packed pages.
void StopBackgroundThread()
Stop the I/O background thread. No-op if already stopped. Called by the destructor.
An in-memory subset of the packed and compressed pages of a cluster.
Definition RCluster.hxx:147
std::unordered_set< ROOT::DescriptorId_t > ColumnSet_t
Definition RCluster.hxx:149
A column element encapsulates the translation between basic C++ types and their column representation...
A column is a storage-backed array of a simple, fixed-size type, from which pages can be mapped into ...
Definition RColumn.hxx:37
A helper class for piece-wise construction of an RNTupleDescriptor.
const RNTupleDescriptor & GetDescriptor() const
The serialization context is used for the piecewise serialization of a descriptor.
std::map< Int_t, TVirtualStreamerInfo * > StreamerInfoMap_t
@ kForReading
Deserializes the descriptor and performs fixup on the suppressed column ranges and on clusters,...
@ kRaw
Deserializes the descriptor as-is without performing any additional fixup.
A memory region that contains packed and compressed pages.
Definition RCluster.hxx:98
Abstract interface to allocate and release pages.
Base class for a sink with a physical storage backend.
RStagedCluster StageCluster(ROOT::NTupleSize_t nNewEntries) final
Stage the current cluster and create a new one for the following data.
void UpdateSchema(const ROOT::Internal::RNTupleModelChangeset &changeset, ROOT::NTupleSize_t firstEntry) override
Incorporate incremental changes to the model into the ntuple descriptor.
void CommitSealedPage(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage) final
Write a preprocessed page to storage. The column must have been added before.
std::uint64_t fNextClusterInGroup
Remembers the starting cluster id for the next cluster group.
std::unique_ptr< RNTupleModel > InitFromDescriptor(const ROOT::RNTupleDescriptor &descriptor, bool copyClusters)
Initialize sink based on an existing descriptor and fill into the descriptor builder,...
void UpdateExtraTypeInfo(const ROOT::RExtraTypeInfoDescriptor &extraTypeInfo) final
Adds an extra type information record to schema.
void CommitAttributeSet(std::string_view attrSetName, const RNTupleLink &attrAnchorInfo) final
Adds the given anchor information (name + locator) into the main RNTuple's descriptor as an attribute...
ColumnHandle_t AddColumn(ROOT::DescriptorId_t fieldId, ROOT::Internal::RColumn &column) final
Register a new column.
ROOT::Internal::RNTupleDescriptorBuilder fDescriptorBuilder
virtual std::vector< RNTupleLocator > CommitSealedPageVImpl(std::span< RPageStorage::RSealedPageGroup > ranges, const std::vector< bool > &mask)
Vector commit of preprocessed pages.
ROOT::NTupleSize_t GetNEntries() const final
void CommitSuppressedColumn(ColumnHandle_t columnHandle) final
Commits a suppressed column for the current cluster.
void AddAliasColumn(const ROOT::RNTupleDescriptor &desc, const ROOT::RFieldDescriptor &field, ROOT::DescriptorId_t physicalId)
Adds a new alias column pointing to an existing column with the given physical id to the given field.
ROOT::Internal::RNTupleSerializer::RContext fSerializationContext
Used to map the IDs of the descriptor to the physical IDs issued during header/footer serialization.
ROOT::NTupleSize_t fPrevClusterNEntries
Used to calculate the number of entries in the current cluster.
virtual RNTupleLink CommitDatasetImpl(unsigned char *serializedFooter, std::uint32_t length)=0
virtual RNTupleLocator CommitClusterGroupImpl(unsigned char *serializedPageList, std::uint32_t length)=0
Returns the locator of the page list envelope of the given buffer that contains the serialized page l...
void CommitStagedClusters(std::span< RStagedCluster > clusters) final
Commit staged clusters, logically appending them to the ntuple descriptor.
static std::unique_ptr< RPageSink > Create(std::string_view ntupleName, std::string_view location, const ROOT::RNTupleWriteOptions &options=ROOT::RNTupleWriteOptions())
Guess the concrete derived page source from the location.
std::vector< ROOT::RClusterDescriptor::RPageRange > fOpenPageRanges
Keeps track of the written pages in the currently open cluster. Indexed by column id.
void CommitPage(ColumnHandle_t columnHandle, const ROOT::Internal::RPage &page) final
Write a page to the storage. The column must have been added before.
virtual void InitImpl(unsigned char *serializedHeader, std::uint32_t length)=0
const ROOT::RNTupleDescriptor & GetDescriptor() const final
Return the RNTupleDescriptor being constructed.
std::vector< ROOT::RClusterDescriptor::RColumnRange > fOpenColumnRanges
Keeps track of the number of elements in the currently open cluster. Indexed by column id.
std::unique_ptr< RCounters > fCounters
void CommitClusterGroup() final
Write out the page locations (page list envelope) for all the committed clusters since the last call ...
virtual RNTupleLocator CommitSealedPageImpl(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage)=0
void CommitSealedPageV(std::span< RPageStorage::RSealedPageGroup > ranges) final
Write a vector of preprocessed pages to storage. The corresponding columns must have been added befor...
void EnableDefaultMetrics(const std::string &prefix)
Enables the default set of metrics provided by RPageSink.
ROOT::DescriptorId_t AddColumnRepresentation(const ROOT::RFieldDescriptor &field, std::span< const ROOT::Internal::RColumnFormat > newRepresentation, std::uint64_t clusterOffset)
Adds a new column representation to the given field.
ROOT::Internal::RNTupleSerializer::StreamerInfoMap_t fInfosOfStreamerFields
Union of the streamer info records that are sent from streamer fields to the sink before committing t...
virtual std::uint64_t StageClusterImpl()=0
Returns the number of bytes written to storage (excluding metadata)
A thread-safe cache of pages loaded from the page source.
Definition RPagePool.hxx:48
Reference to a page stored in the page pool.
An RAII wrapper used to synchronize a page sink. See GetSinkGuard().
RSinkGuard & operator=(const RSinkGuard &)=delete
RSinkGuard & operator=(RSinkGuard &&)=delete
RSinkGuard(const RSinkGuard &)=delete
Abstract interface to write data into an ntuple.
virtual RNTupleLink CommitDatasetImpl()=0
RPageSink(RPageSink &&)=default
RNTupleLink CommitDataset()
Run the registered callbacks and finalize the current cluster and the entrire data set.
virtual void CommitSealedPage(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage)=0
Write a preprocessed page to storage. The column must have been added before.
std::vector< unsigned char > fSealPageBuffer
Used as destination buffer in the simple SealPage overload.
virtual void CommitSuppressedColumn(ColumnHandle_t columnHandle)=0
Commits a suppressed column for the current cluster.
std::unique_ptr< ROOT::RNTupleWriteOptions > fOptions
std::vector< Callback_t > fOnDatasetCommitCallbacks
virtual ROOT::Internal::RPage ReservePage(ColumnHandle_t columnHandle, std::size_t nElements)
Get a new, empty page for the given column that can be filled with up to nElements; nElements must be...
const ROOT::RNTupleWriteOptions & GetWriteOptions() const
Returns the sink's write options.
virtual void CommitClusterGroup()=0
Write out the page locations (page list envelope) for all the committed clusters since the last call ...
virtual RStagedCluster StageCluster(ROOT::NTupleSize_t nNewEntries)=0
Stage the current cluster and create a new one for the following data.
virtual void CommitPage(ColumnHandle_t columnHandle, const ROOT::Internal::RPage &page)=0
Write a page to the storage. The column must have been added before.
RPageSink & operator=(const RPageSink &)=delete
EPageStorageType GetType() final
Whether the concrete implementation is a sink or a source.
void Init(RNTupleModel &model)
Physically creates the storage container to hold the ntuple (e.g., a keys a TFile or an S3 bucket) In...
virtual void UpdateExtraTypeInfo(const ROOT::RExtraTypeInfoDescriptor &extraTypeInfo)=0
Adds an extra type information record to schema.
virtual void CommitAttributeSet(std::string_view attrSetName, const RNTupleLink &attrAnchorInfo)=0
Adds the given anchor information (name + locator) into the main RNTuple's descriptor as an attribute...
RPageSink(const RPageSink &)=delete
virtual std::unique_ptr< RPageSink > CloneAsHidden(std::string_view name, const RNTupleWriteOptions &opts) const =0
Creates a new sink with the same underlying storage as this but writing to a different RNTuple named ...
virtual void CommitStagedClusters(std::span< RStagedCluster > clusters)=0
Commit staged clusters, logically appending them to the ntuple descriptor.
virtual std::uint64_t CommitCluster(ROOT::NTupleSize_t nNewEntries)
Finalize the current cluster and create a new one for the following data.
virtual void InitImpl(RNTupleModel &model)=0
virtual RSinkGuard GetSinkGuard()
std::function< void(RPageSink &)> Callback_t
virtual void UpdateSchema(const ROOT::Internal::RNTupleModelChangeset &changeset, ROOT::NTupleSize_t firstEntry)=0
Incorporate incremental changes to the model into the ntuple descriptor.
void RegisterOnCommitDatasetCallback(const Callback_t &cb)
The registered callback is executed at the beginning of CommitDataset();.
RSealedPage SealPage(const ROOT::Internal::RPage &page, const ROOT::Internal::RColumnElementBase &element)
Helper for streaming a page.
bool fIsInitialized
Flag if sink was initialized.
RPageSink(std::string_view ntupleName, const ROOT::RNTupleWriteOptions &options)
RPageSink & operator=(RPageSink &&)=default
virtual const ROOT::RNTupleDescriptor & GetDescriptor() const =0
Return the RNTupleDescriptor being constructed.
virtual void CommitSealedPageV(std::span< RPageStorage::RSealedPageGroup > ranges)=0
Write a vector of preprocessed pages to storage. The corresponding columns must have been added befor...
virtual ROOT::NTupleSize_t GetNEntries() const =0
RWritePageMemoryManager fWritePageMemoryManager
Used in ReservePage to maintain the page buffer budget.
void DropColumn(ColumnHandle_t) final
Unregisters a column.
Keeps track of the requested physical column IDs and their in-memory target type via a column element...
const std::vector< RColumnInfo > & GetColumnInfos(ROOT::DescriptorId_t physicalColumnId) const
std::unordered_map< ROOT::DescriptorId_t, std::vector< RColumnInfo > > fColumnInfos
Maps physical column IDs to all the requested in-memory representations.
bool HasColumnInfos(ROOT::DescriptorId_t physicalColumnId) const
Either a shared or an exclusive lock guard.
RAnyDescriptorGuard(RAnyDescriptorGuard &&)=default
RAnyDescriptorGuard(RExclDescriptorGuard exclGuard)
std::variant< RSharedDescriptorGuard, RExclDescriptorGuard > fAnyGuard
const ROOT::RNTupleDescriptor & GetConstRef() const
RAnyDescriptorGuard & operator=(RAnyDescriptorGuard &&)=default
const ROOT::RNTupleDescriptor * operator->() const
RAnyDescriptorGuard(RSharedDescriptorGuard sharedGuard)
RAnyDescriptorGuard(const RAnyDescriptorGuard &)=delete
RAnyDescriptorGuard & operator=(const RAnyDescriptorGuard &)=delete
An RAII wrapper used for the writable access to RPageSource::fDescriptor. See GetSharedDescriptorGuar...
ROOT::RNTupleDescriptor & operator*() const
void MoveIn(ROOT::RNTupleDescriptor desc)
RExclDescriptorGuard & operator=(const RExclDescriptorGuard &)=delete
ROOT::RNTupleDescriptor * operator->() const
RExclDescriptorGuard(RExclDescriptorGuard &&other)
RExclDescriptorGuard(ROOT::RNTupleDescriptor &desc, std::shared_mutex &lock)
RExclDescriptorGuard(const RExclDescriptorGuard &)=delete
RExclDescriptorGuard & operator=(RExclDescriptorGuard &&other)
An RAII wrapper used for the read-only access to RPageSource::fDescriptor. See GetExclDescriptorGuard...
RSharedDescriptorGuard(const RSharedDescriptorGuard &)=delete
const ROOT::RNTupleDescriptor & GetConstRef() const
RSharedDescriptorGuard(RSharedDescriptorGuard &&other)
RSharedDescriptorGuard & operator=(const RSharedDescriptorGuard &)=delete
const ROOT::RNTupleDescriptor * operator->() const
RSharedDescriptorGuard(const ROOT::RNTupleDescriptor &desc, std::shared_mutex &lock)
RSharedDescriptorGuard & operator=(RSharedDescriptorGuard &&other)
Abstract interface to read data from an ntuple.
ROOT::Internal::RClusterPool fClusterPool
The cluster pool asynchronously preloads the next few clusters.
const std::unordered_set< ROOT::DescriptorId_t > & GetPinnedClusters() const
RPageSource & operator=(RPageSource &&)=delete
EPageStorageType GetType() final
Whether the concrete implementation is a sink or a source.
RPageSource & operator=(const RPageSource &)=delete
RExclDescriptorGuard GetExclDescriptorGuard()
Note that the underlying lock is not recursive. See GetSharedDescriptorGuard() for further informatio...
ROOT::RNTupleReadOptions fOptions
virtual void LoadPageListImpl(const RNTupleLocator &locator, unsigned char *buffer)=0
REntryRange GetEntryRange() const
std::map< ROOT::NTupleSize_t, ROOT::DescriptorId_t > fPreloadedClusters
Clusters from where pages got preloaded in UnzipCluster(), ordered by first entry number of the clust...
RSharedDescriptorGuard GetSharedDescriptorGuard() const
Takes the read lock for the descriptor.
REntryRange fEntryRange
Used by the cluster pool to prevent reading beyond the given range.
RStructureBuffer fStructureBuffer
Populated by LoadStructureImpl(), reset at the end of Attach()
virtual std::unique_ptr< RPageSource > CloneImpl() const =0
Returns a new, unattached page source for the same data set.
void UnpinCluster(ROOT::DescriptorId_t clusterId)
Allows the given cluster to be evicted from the cluster pool and page pool.
virtual void LoadSealedPageImpl(const RNTupleLocator &locator, RSealedPage &sealedPage)=0
std::unique_ptr< RCounters > fCounters
std::unordered_set< ROOT::DescriptorId_t > fPinnedClusters
Pinned clusters and their $2 * (cluster bunch size) - 1$ successors will not be evicted from the clus...
virtual void LoadStructureImpl()=0
Fills fStructureBuffer with the compressed header and footer.
virtual ROOT::RNTupleDescriptor AttachImpl()=0
LoadStructureImpl() has been called before AttachImpl() is called
ROOT::RNTupleDescriptor fDescriptor
void PinCluster(ROOT::DescriptorId_t clusterId)
Instructs the cluster pool and page pool to consider the given cluster as active (should stay cached)...
ROOT::Internal::RPagePool fPagePool
Pages that are unzipped with IMT are staged into the page pool.
std::vector< NTupleSize_t > fCumulativeClusterCounts
This vector is aligned with the cluster groups in the descriptor.
std::shared_mutex fDescriptorLock
virtual std::vector< std::unique_ptr< ROOT::Internal::RCluster > > LoadClusters(std::span< ROOT::Internal::RCluster::RKey > clusterKeys)=0
Populates all the pages of the given cluster ids and columns; it is possible that some columns do not...
RPageSource(RPageSource &&)=delete
virtual std::unique_ptr< RPageSource > OpenWithDifferentAnchor(const ROOT::Internal::RNTupleLink &anchorLink, const ROOT::RNTupleReadOptions &options={})=0
Creates a new PageSource using the same underlying file as this but referring to a different RNTuple,...
RPageSource(const RPageSource &)=delete
const ROOT::RNTupleReadOptions & GetReadOptions() const
RActivePhysicalColumns fActivePhysicalColumns
The active columns are implicitly defined by the model fields or views.
The interface of a task scheduler to schedule page (de)compression tasks.
virtual void Wait()=0
Blocks until all scheduled tasks finished.
virtual void AddTask(const std::function< void(void)> &taskFunc)=0
Take a callable that represents a task.
Common functionality of an ntuple storage for both reading and writing.
virtual ColumnHandle_t AddColumn(ROOT::DescriptorId_t fieldId, ROOT::Internal::RColumn &column)=0
Register a new column.
void SetTaskScheduler(RTaskScheduler *taskScheduler)
RPageStorage(const RPageStorage &other)=delete
virtual ROOT::Experimental::Detail::RNTupleMetrics & GetMetrics()
Returns the default metrics object.
RPageStorage & operator=(const RPageStorage &other)=delete
ROOT::DescriptorId_t GetColumnId(ColumnHandle_t columnHandle) const
virtual void DropColumn(ColumnHandle_t columnHandle)=0
Unregisters a column.
RPageStorage & operator=(RPageStorage &&other)=default
std::unique_ptr< ROOT::Internal::RPageAllocator > fPageAllocator
For the time being, we will use the heap allocator for all sources and sinks. This may change in the ...
static constexpr std::size_t kNBytesPageChecksum
The page checksum is a 64bit xxhash3.
std::deque< RSealedPage > SealedPageSequence_t
virtual EPageStorageType GetType()=0
Whether the concrete implementation is a sink or a source.
RColumnHandle ColumnHandle_t
The column handle identifies a column with the current open page storage.
RPageStorage(RPageStorage &&other)=default
ROOT::Experimental::Detail::RNTupleMetrics fMetrics
RPageStorage(std::string_view name)
const std::string & GetNTupleName() const
Returns the NTuple name.
A page is a slice of a column that is mapped into memory.
Definition RPage.hxx:43
Helper to maintain a memory budget for the write pages of a set of columns.
std::size_t fCurrentAllocatedBytes
Sum of all the write page sizes (their capacity) of the columns in fColumnsSortedByPageSize
std::size_t fMaxAllocatedBytes
Maximum allowed value for fCurrentAllocatedBytes, set from RNTupleWriteOptions::fPageBufferBudget.
bool TryEvict(std::size_t targetAvailableSize, std::size_t pageSizeLimit)
Flush columns in order of allocated write page size until the sum of all write page allocations leave...
bool TryUpdate(ROOT::Internal::RColumn &column, std::size_t newWritePageSize)
Try to register the new write page size for the given column.
RWritePageMemoryManager(std::size_t maxAllocatedBytes)
std::set< RColumnInfo, std::greater< RColumnInfo > > fColumnsSortedByPageSize
All columns that called ReservePage() (hence TryUpdate()) at least once, sorted by their current writ...
Records the partition of data into pages for a particular column in a particular cluster.
Metadata for RNTuple clusters.
Base class for all ROOT issued exceptions.
Definition RError.hxx:78
Field specific extra type information from the header / extenstion header.
Metadata stored for every field of an RNTuple.
The on-storage metadata of an RNTuple.
Addresses a column element or field item relative to a particular cluster, instead of a global NTuple...
Generic information about the physical location of data.
The RNTupleModel encapulates the schema of an RNTuple.
Common user-tunable settings for reading RNTuples.
Common user-tunable settings for storing RNTuples.
The class is used as a return type for operations that can fail; wraps a value of type T or an RError...
Definition RError.hxx:222
std::uint64_t DescriptorId_t
Distriniguishes elements of the same type within a descriptor, e.g. different fields.
constexpr NTupleSize_t kInvalidNTupleIndex
std::uint64_t NTupleSize_t
Integer type long enough to hold the maximum number of entries in a column.
constexpr DescriptorId_t kInvalidDescriptorId
The identifiers that specifies the content of a (partial) cluster.
Definition RCluster.hxx:151
Every concrete RColumnElement type is identified by its on-disk type (column type) and the in-memory ...
The incremental changes to a RNTupleModel
Default I/O performance counters that get registered in fMetrics.
ROOT::Experimental::Detail::RNTupleAtomicCounter & fSzWritePayload
ROOT::Experimental::Detail::RNTupleTickCounter< ROOT::Experimental::Detail::RNTupleAtomicCounter > & fTimeCpuWrite
ROOT::Experimental::Detail::RNTupleAtomicCounter & fTimeWallWrite
ROOT::Experimental::Detail::RNTupleTickCounter< ROOT::Experimental::Detail::RNTupleAtomicCounter > & fTimeCpuZip
ROOT::Experimental::Detail::RNTupleAtomicCounter & fTimeWallZip
ROOT::Experimental::Detail::RNTupleAtomicCounter & fSzZip
ROOT::Experimental::Detail::RNTupleAtomicCounter & fNPageCommitted
Set of optional features supported by the persistent sink.
Parameters for the SealPage() method.
bool fWriteChecksum
Adds a 8 byte little-endian xxhash3 checksum to the page payload.
std::uint32_t fCompressionSettings
Compression algorithm and level to apply.
void * fBuffer
Location for sealed output. The memory buffer has to be large enough.
const ROOT::Internal::RPage * fPage
Input page to be sealed.
bool fAllowAlias
If false, the output buffer must not point to the input page buffer, which would otherwise be an opti...
const ROOT::Internal::RColumnElementBase * fElement
Corresponds to the page's elements, for size calculation etc.
ROOT::RClusterDescriptor::RPageRange fPageRange
Cluster that was staged, but not yet logically appended to the RNTuple.
std::vector< RColumnInfo > fColumnInfos
ROOT::Internal::RColumnElementBase::RIdentifier fElementId
Default I/O performance counters that get registered in fMetrics
ROOT::Experimental::Detail::RNTupleAtomicCounter & fNPageUnsealed
ROOT::Experimental::Detail::RNTupleAtomicCounter & fNRead
ROOT::Experimental::Detail::RNTupleAtomicCounter & fTimeWallRead
ROOT::Experimental::Detail::RNTupleAtomicCounter & fSzReadOverhead
ROOT::Experimental::Detail::RNTupleCalcPerf & fCompressionRatio
ROOT::Experimental::Detail::RNTupleCalcPerf & fFractionReadOverhead
ROOT::Experimental::Detail::RNTupleAtomicCounter & fNReadV
ROOT::Experimental::Detail::RNTupleAtomicCounter & fNClusterLoaded
ROOT::Experimental::Detail::RNTupleAtomicCounter & fTimeWallUnzip
ROOT::Experimental::Detail::RNTupleAtomicCounter & fNPageRead
ROOT::Experimental::Detail::RNTupleAtomicCounter & fSzReadPayload
ROOT::Experimental::Detail::RNTupleCalcPerf & fBandwidthReadCompressed
ROOT::Experimental::Detail::RNTupleTickCounter< ROOT::Experimental::Detail::RNTupleAtomicCounter > & fTimeCpuUnzip
ROOT::Experimental::Detail::RNTupleTickCounter< ROOT::Experimental::Detail::RNTupleAtomicCounter > & fTimeCpuRead
ROOT::Experimental::Detail::RNTupleCalcPerf & fBandwidthReadUncompressed
ROOT::Experimental::Detail::RNTupleCalcPerf & fBandwidthUnzip
ROOT::Experimental::Detail::RNTupleAtomicCounter & fSzUnzip
Used in SetEntryRange / GetEntryRange.
Summarizes meta-data necessary to load a certain page. Used by LoadPageFromSummary().
ROOT::RClusterDescriptor::RPageInfoExtended fPageInfo
Location of the page on disk.
Holds the uncompressed header and footer.
void Reset()
Called at the end of Attach(), i.e. when the header and footer are processed.
std::unique_ptr< unsigned char[]> fBuffer
single buffer for both header and footer
A range of sealed pages referring to the same column that can be used for vector commit.
SealedPageSequence_t::const_iterator fFirst
SealedPageSequence_t::const_iterator fLast
RSealedPageGroup(ROOT::DescriptorId_t d, const SealedPageSequence_t::const_iterator &b, const SealedPageSequence_t::const_iterator &e)
A sealed page contains the bytes of a page as written to storage (packed & compressed).
RSealedPage(const void *buffer, std::size_t bufferSize, std::uint32_t nElements, bool hasChecksum=false)
RResult< void > VerifyChecksumIfEnabled() const
RSealedPage(const RSealedPage &other)=default
bool fHasChecksum
If set, the last 8 bytes of the buffer are the xxhash of the rest of the buffer.
RSealedPage & operator=(RSealedPage &&other)=default
void SetNElements(std::uint32_t nElements)
RSealedPage(RSealedPage &&other)=default
RResult< std::uint64_t > GetChecksum() const
Returns a failure if the sealed page has no checksum.
RSealedPage & operator=(const RSealedPage &other)=default
void SetBufferSize(std::size_t bufferSize)
std::size_t fBufferSize
Size of the page payload and the trailing checksum (if available)
bool operator>(const RColumnInfo &other) const
Additional information about a page in an in-memory RPageRange.
Information about a single page in the context of a cluster's page range.