RColumn.hxx

Go to the documentation of this file.

/// \file ROOT/RColumn.hxx
/// \ingroup NTuple ROOT7
/// \author Jakob Blomer <jblomer@cern.ch>
/// \date 2018-10-09
/// \warning This is part of the ROOT 7 prototype! It will change without notice. It might trigger earthquakes. Feedback
/// is welcome!
 
/*************************************************************************
 * Copyright (C) 1995-2019, Rene Brun and Fons Rademakers.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
 *************************************************************************/
 
#ifndef ROOT7_RColumn
#define ROOT7_RColumn
 
#include <ROOT/RConfig.hxx> // for R__likely
#include <ROOT/RColumnElementBase.hxx>
#include <ROOT/RNTupleUtil.hxx>
#include <ROOT/RPage.hxx>
#include <ROOT/RPageStorage.hxx>
 
#include <TError.h>
 
#include <cstring> // for memcpy
#include <memory>
#include <utility>
 
namespace ROOT {
namespace Experimental {
namespace Internal {
 
// clang-format off
/**
\class ROOT::Experimental::Internal::RColumn
\ingroup NTuple
\brief A column is a storage-backed array of a simple, fixed-size type, from which pages can be mapped into memory.
*/
// clang-format on
class RColumn {
private:
   EColumnType fType;
   std::uint16_t fBitsOnStorage = 0;
   /// Columns belonging to the same field are distinguished by their order.  E.g. for an std::string field, there is
   /// the offset column with index 0 and the character value column with index 1.
   std::uint32_t fIndex;
   /// Fields can have multiple column representations, distinguished by representation index
   std::uint16_t fRepresentationIndex;
   RPageSink *fPageSink = nullptr;
   RPageSource *fPageSource = nullptr;
   RPageStorage::ColumnHandle_t fHandleSink;
   RPageStorage::ColumnHandle_t fHandleSource;
   /// A set of open pages into which new elements are being written. The pages are used
   /// in rotation. If tail page optimization is enabled, they are 50% bigger than the target size
   /// given by the write options. The current page is filled until the target size, but it is only
   /// committed once the other write page is filled at least 50%. If a flush occurs earlier, a
   /// slightly oversized, single page will be committed.
   /// Without tail page optimization, only one page is allocated equal to the target size.
   RPage fWritePage[2];
   /// Index of the current write page
   int fWritePageIdx = 0;
   /// For writing, the targeted number of elements, given by `fApproxNElementsPerPage` (in the write options) and the
   /// element size. We ensure this value to be >= 2 in Connect() so that we have meaningful "page full" and "page half
   /// full" events when writing the page.
   std::uint32_t fApproxNElementsPerPage = 0;
   /// The number of elements written resp. available in the column
   NTupleSize_t fNElements = 0;
   /// The currently mapped page for reading
   RPageRef fReadPageRef;
   /// The column id is used to find matching pages with content when reading
   ColumnId_t fColumnIdSource = kInvalidColumnId;
   /// Global index of the first element in this column; usually == 0, unless it is a deferred column
   NTupleSize_t fFirstElementIndex = 0;
   /// Used to pack and unpack pages on writing/reading
   std::unique_ptr<RColumnElementBase> fElement;
   /// The column team is a set of columns that serve the same column index for different representation IDs.
   /// Initially, the team has only one member, the very column it belongs to. Through MergeTeams(), two columns
   /// can join forces. The team is used to react on suppressed columns: if the current team member has a suppressed
   /// column for a MapPage() call, it get the page from the active column in the corresponding cluster.
   std::vector<RColumn *> fTeam;
   /// Points into fTeam to the column that successfully returned the last page.
   std::size_t fLastGoodTeamIdx = 0;
 
   RColumn(EColumnType type, std::uint32_t columnIndex, std::uint16_t representationIndex);
 
   /// Used in Append() and AppendV() to handle the case when the main page reached the target size.
   /// If tail page optimization is enabled, switch the pages; the other page has been flushed when
   /// the main page reached 50%.
   /// Without tail page optimization, flush the current page to make room for future writes.
   void HandleWritePageIfFull()
   {
      if (R__likely(fWritePage[fWritePageIdx].GetNElements() < fApproxNElementsPerPage))
         return;
 
      auto otherIdx = 1 - fWritePageIdx; // == (fWritePageIdx + 1) % 2
      if (fWritePage[otherIdx].IsNull()) {
         // There is only this page; we have to flush it now to make room for future writes.
         fPageSink->CommitPage(fHandleSink, fWritePage[fWritePageIdx]);
         fWritePage[fWritePageIdx].Reset(fNElements);
      } else {
         fWritePageIdx = otherIdx;
         R__ASSERT(fWritePage[fWritePageIdx].IsEmpty());
         fWritePage[fWritePageIdx].Reset(fNElements);
      }
   }
 
   /// When the main write page surpasses the 50% fill level, the (full) shadow write page gets flushed
   void FlushShadowWritePage()
   {
      auto otherIdx = 1 - fWritePageIdx;
      if (fWritePage[otherIdx].IsEmpty())
         return;
      fPageSink->CommitPage(fHandleSink, fWritePage[otherIdx]);
      // Mark the page as flushed; the rangeFirst is zero for now but will be reset to
      // fNElements in SwapWritePagesIfFull() when the pages swap
      fWritePage[otherIdx].Reset(0);
   }
 
public:
   template <typename CppT>
   static std::unique_ptr<RColumn> Create(EColumnType type, std::uint32_t columnIdx, std::uint16_t representationIdx)
   {
      auto column = std::unique_ptr<RColumn>(new RColumn(type, columnIdx, representationIdx));
      column->fElement = RColumnElementBase::Generate<CppT>(type);
      return column;
   }
 
   RColumn(const RColumn &) = delete;
   RColumn &operator=(const RColumn &) = delete;
   ~RColumn();
 
   /// Connect the column to a page sink.  `firstElementIndex` can be used to specify the first column element index
   /// with backing storage for this column.  On read back, elements before `firstElementIndex` will cause the zero page
   /// to be mapped.
   void ConnectPageSink(DescriptorId_t fieldId, RPageSink &pageSink, NTupleSize_t firstElementIndex = 0U);
   /// Connect the column to a page source.
   void ConnectPageSource(DescriptorId_t fieldId, RPageSource &pageSource);
 
   void Append(const void *from)
   {
      void *dst = fWritePage[fWritePageIdx].GrowUnchecked(1);
 
      if (fWritePage[fWritePageIdx].GetNElements() == fApproxNElementsPerPage / 2) {
         FlushShadowWritePage();
      }
 
      std::memcpy(dst, from, fElement->GetSize());
      fNElements++;
 
      HandleWritePageIfFull();
   }
 
   void AppendV(const void *from, std::size_t count)
   {
      // We might not have enough space in the current page. In this case, fall back to one by one filling.
      if (fWritePage[fWritePageIdx].GetNElements() + count > fApproxNElementsPerPage) {
         // TODO(jblomer): use (fewer) calls to AppendV to write the data page-by-page
         for (unsigned i = 0; i < count; ++i) {
            Append(static_cast<const unsigned char *>(from) + fElement->GetSize() * i);
         }
         return;
      }
 
      // The check for flushing the shadow page is more complicated than for the Append() case
      // because we don't necessarily fill up to exactly fApproxNElementsPerPage / 2 elements;
      // we might instead jump over the 50% fill level.
      // This check should be done before calling `RPage::GrowUnchecked()` as the latter affects the return value of
      // `RPage::GetNElements()`.
      if ((fWritePage[fWritePageIdx].GetNElements() < fApproxNElementsPerPage / 2) &&
          (fWritePage[fWritePageIdx].GetNElements() + count >= fApproxNElementsPerPage / 2)) {
         FlushShadowWritePage();
      }
 
      void *dst = fWritePage[fWritePageIdx].GrowUnchecked(count);
 
      std::memcpy(dst, from, fElement->GetSize() * count);
      fNElements += count;
 
      // Note that by the very first check in AppendV, we cannot have filled more than fApproxNElementsPerPage elements
      HandleWritePageIfFull();
   }
 
   void Read(const NTupleSize_t globalIndex, void *to)
   {
      if (!fReadPageRef.Get().Contains(globalIndex)) {
         MapPage(globalIndex);
      }
      const auto elemSize = fElement->GetSize();
      void *from = static_cast<unsigned char *>(fReadPageRef.Get().GetBuffer()) +
                   (globalIndex - fReadPageRef.Get().GetGlobalRangeFirst()) * elemSize;
      std::memcpy(to, from, elemSize);
   }
 
   void Read(RClusterIndex clusterIndex, void *to)
   {
      if (!fReadPageRef.Get().Contains(clusterIndex)) {
         MapPage(clusterIndex);
      }
      const auto elemSize = fElement->GetSize();
      void *from = static_cast<unsigned char *>(fReadPageRef.Get().GetBuffer()) +
                   (clusterIndex.GetIndex() - fReadPageRef.Get().GetClusterRangeFirst()) * elemSize;
      std::memcpy(to, from, elemSize);
   }
 
   void ReadV(const NTupleSize_t globalIndex, const ClusterSize_t::ValueType count, void *to)
   {
      if (!fReadPageRef.Get().Contains(globalIndex)) {
         MapPage(globalIndex);
      }
      NTupleSize_t idxInPage = globalIndex - fReadPageRef.Get().GetGlobalRangeFirst();
 
      const auto elemSize = fElement->GetSize();
      const void *from = static_cast<unsigned char *>(fReadPageRef.Get().GetBuffer()) + idxInPage * elemSize;
      if (globalIndex + count <= fReadPageRef.Get().GetGlobalRangeLast() + 1) {
         std::memcpy(to, from, elemSize * count);
      } else {
         ClusterSize_t::ValueType nBatch = fReadPageRef.Get().GetNElements() - idxInPage;
         std::memcpy(to, from, elemSize * nBatch);
         auto tail = static_cast<unsigned char *>(to) + nBatch * elemSize;
         ReadV(globalIndex + nBatch, count - nBatch, tail);
      }
   }
 
   void ReadV(RClusterIndex clusterIndex, const ClusterSize_t::ValueType count, void *to)
   {
      if (!fReadPageRef.Get().Contains(clusterIndex)) {
         MapPage(clusterIndex);
      }
      NTupleSize_t idxInPage = clusterIndex.GetIndex() - fReadPageRef.Get().GetClusterRangeFirst();
 
      const auto elemSize = fElement->GetSize();
      const void *from = static_cast<unsigned char *>(fReadPageRef.Get().GetBuffer()) + idxInPage * elemSize;
      if (clusterIndex.GetIndex() + count <= fReadPageRef.Get().GetClusterRangeLast() + 1) {
         std::memcpy(to, from, elemSize * count);
      } else {
         ClusterSize_t::ValueType nBatch = fReadPageRef.Get().GetNElements() - idxInPage;
         std::memcpy(to, from, elemSize * nBatch);
         auto tail = static_cast<unsigned char *>(to) + nBatch * elemSize;
         ReadV(RClusterIndex(clusterIndex.GetClusterId(), clusterIndex.GetIndex() + nBatch), count - nBatch, tail);
      }
   }
 
   template <typename CppT>
   CppT *Map(const NTupleSize_t globalIndex)
   {
      NTupleSize_t nItems;
      return MapV<CppT>(globalIndex, nItems);
   }
 
   template <typename CppT>
   CppT *Map(RClusterIndex clusterIndex)
   {
      NTupleSize_t nItems;
      return MapV<CppT>(clusterIndex, nItems);
   }
 
   template <typename CppT>
   CppT *MapV(const NTupleSize_t globalIndex, NTupleSize_t &nItems)
   {
      if (R__unlikely(!fReadPageRef.Get().Contains(globalIndex))) {
         MapPage(globalIndex);
      }
      // +1 to go from 0-based indexing to 1-based number of items
      nItems = fReadPageRef.Get().GetGlobalRangeLast() - globalIndex + 1;
      return reinterpret_cast<CppT *>(static_cast<unsigned char *>(fReadPageRef.Get().GetBuffer()) +
                                      (globalIndex - fReadPageRef.Get().GetGlobalRangeFirst()) * sizeof(CppT));
   }
 
   template <typename CppT>
   CppT *MapV(RClusterIndex clusterIndex, NTupleSize_t &nItems)
   {
      if (!fReadPageRef.Get().Contains(clusterIndex)) {
         MapPage(clusterIndex);
      }
      // +1 to go from 0-based indexing to 1-based number of items
      nItems = fReadPageRef.Get().GetClusterRangeLast() - clusterIndex.GetIndex() + 1;
      return reinterpret_cast<CppT *>(static_cast<unsigned char *>(fReadPageRef.Get().GetBuffer()) +
                                      (clusterIndex.GetIndex() - fReadPageRef.Get().GetClusterRangeFirst()) *
                                         sizeof(CppT));
   }
 
   NTupleSize_t GetGlobalIndex(RClusterIndex clusterIndex)
   {
      if (!fReadPageRef.Get().Contains(clusterIndex)) {
         MapPage(clusterIndex);
      }
      return fReadPageRef.Get().GetClusterInfo().GetIndexOffset() + clusterIndex.GetIndex();
   }
 
   RClusterIndex GetClusterIndex(NTupleSize_t globalIndex)
   {
      if (!fReadPageRef.Get().Contains(globalIndex)) {
         MapPage(globalIndex);
      }
      return RClusterIndex(fReadPageRef.Get().GetClusterInfo().GetId(),
                           globalIndex - fReadPageRef.Get().GetClusterInfo().GetIndexOffset());
   }
 
   /// For offset columns only, look at the two adjacent values that define a collection's coordinates
   void GetCollectionInfo(const NTupleSize_t globalIndex, RClusterIndex *collectionStart, ClusterSize_t *collectionSize)
   {
      NTupleSize_t idxStart = 0;
      NTupleSize_t idxEnd;
      // Try to avoid jumping back to the previous page and jumping back to the previous cluster
      if (R__likely(globalIndex > 0)) {
         if (R__likely(fReadPageRef.Get().Contains(globalIndex - 1))) {
            idxStart = *Map<ClusterSize_t>(globalIndex - 1);
            idxEnd = *Map<ClusterSize_t>(globalIndex);
            if (R__unlikely(fReadPageRef.Get().GetClusterInfo().GetIndexOffset() == globalIndex))
               idxStart = 0;
         } else {
            idxEnd = *Map<ClusterSize_t>(globalIndex);
            auto selfOffset = fReadPageRef.Get().GetClusterInfo().GetIndexOffset();
            idxStart = (globalIndex == selfOffset) ? 0 : *Map<ClusterSize_t>(globalIndex - 1);
         }
      } else {
         idxEnd = *Map<ClusterSize_t>(globalIndex);
      }
      *collectionSize = idxEnd - idxStart;
      *collectionStart = RClusterIndex(fReadPageRef.Get().GetClusterInfo().GetId(), idxStart);
   }
 
   void GetCollectionInfo(RClusterIndex clusterIndex, RClusterIndex *collectionStart, ClusterSize_t *collectionSize)
   {
      auto index = clusterIndex.GetIndex();
      auto idxStart = (index == 0) ? 0 : *Map<ClusterSize_t>(clusterIndex - 1);
      auto idxEnd = *Map<ClusterSize_t>(clusterIndex);
      *collectionSize = idxEnd - idxStart;
      *collectionStart = RClusterIndex(clusterIndex.GetClusterId(), idxStart);
   }
 
   /// Get the currently active cluster id
   void GetSwitchInfo(NTupleSize_t globalIndex, RClusterIndex *varIndex, std::uint32_t *tag)
   {
      auto varSwitch = Map<RColumnSwitch>(globalIndex);
      *varIndex = RClusterIndex(fReadPageRef.Get().GetClusterInfo().GetId(), varSwitch->GetIndex());
      *tag = varSwitch->GetTag();
   }
 
   void Flush();
   void CommitSuppressed();
 
   void MapPage(NTupleSize_t globalIndex) { R__ASSERT(TryMapPage(globalIndex)); }
   void MapPage(RClusterIndex clusterIndex) { R__ASSERT(TryMapPage(clusterIndex)); }
   bool TryMapPage(NTupleSize_t globalIndex);
   bool TryMapPage(RClusterIndex clusterIndex);
 
   bool ReadPageContains(NTupleSize_t globalIndex) const { return fReadPageRef.Get().Contains(globalIndex); }
   bool ReadPageContains(RClusterIndex clusterIndex) const { return fReadPageRef.Get().Contains(clusterIndex); }
 
   void MergeTeams(RColumn &other);
 
   NTupleSize_t GetNElements() const { return fNElements; }
   RColumnElementBase *GetElement() const { return fElement.get(); }
   EColumnType GetType() const { return fType; }
   std::uint16_t GetBitsOnStorage() const { return fBitsOnStorage; }
   std::uint32_t GetIndex() const { return fIndex; }
   std::uint16_t GetRepresentationIndex() const { return fRepresentationIndex; }
   ColumnId_t GetColumnIdSource() const { return fColumnIdSource; }
   NTupleSize_t GetFirstElementIndex() const { return fFirstElementIndex; }
   RPageSource *GetPageSource() const { return fPageSource; }
   RPageSink *GetPageSink() const { return fPageSink; }
   RPageStorage::ColumnHandle_t GetHandleSource() const { return fHandleSource; }
   RPageStorage::ColumnHandle_t GetHandleSink() const { return fHandleSink; }
}; // class RColumn
 
} // namespace Internal
} // namespace Experimental
} // namespace ROOT
 
#endif