[ntuple] Support multiple column representations in the merger

tree/ntuple/inc/ROOT/RFieldBase.hxx

@@ -247,14 +247,15 @@ private:
          func(target);
    }
 
-   /// Translate an entry index to a column element index of the principal column and vice versa. These functions
-   /// take into account the role and number of repetitions on each level of the field hierarchy as follows:
+   /// Translate an entry index to a column element index of the principal column. This function
+   /// takes into account the role and number of repetitions on each level of the field hierarchy as follows:
    /// - Top level fields: element index == entry index
    /// - Record fields propagate their principal column index to the principal columns of direct descendant fields
    /// - Collection and variant fields set the principal column index of their children to 0
    ///
    /// The column element index also depends on the number of repetitions of each field in the hierarchy, e.g., given a
-   /// field with type `std::array<std::array<float, 4>, 2>`, this function returns 8 for the innermost field.
+   /// field with type `std::array<std::array<float, 4>, 2>`, this function called with `globalIndex == 1`
+   /// returns 8 for the innermost field.
    ROOT::NTupleSize_t EntryToColumnElementIndex(ROOT::NTupleSize_t globalIndex) const;
 
    /// Flushes data from active columns

tree/ntuple/inc/ROOT/RNTupleDescriptor.hxx

@@ -519,6 +519,12 @@ public:
    ROOT::NTupleSize_t GetFirstEntryIndex() const { return fFirstEntryIndex; }
    ROOT::NTupleSize_t GetNEntries() const { return fNEntries; }
    const RColumnRange &GetColumnRange(ROOT::DescriptorId_t physicalId) const { return fColumnRanges.at(physicalId); }
+   const RColumnRange *TryGetColumnRange(ROOT::DescriptorId_t physicalId) const
+   {
+      if (auto it = fColumnRanges.find(physicalId); it != fColumnRanges.end())
+         return &it->second;
+      return nullptr;
+   }
    const RPageRange &GetPageRange(ROOT::DescriptorId_t physicalId) const { return fPageRanges.at(physicalId); }
    /// Returns an iterator over pairs { columnId, columnRange }. The iteration order is unspecified.
    RColumnRangeIterable GetColumnRangeIterable() const;

tree/ntuple/inc/ROOT/RNTupleMerger.hxx

@@ -118,16 +118,16 @@ class RNTupleMerger final {
    std::unique_ptr<ROOT::RNTupleModel> fModel;
 
    [[nodiscard]]
-   ROOT::RResult<void> MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool,
-                                          const ROOT::RClusterDescriptor &clusterDesc,
-                                          std::span<const RColumnMergeInfo> commonColumns,
-                                          const ROOT::Internal::RCluster::ColumnSet_t &commonColumnSet,
-                                          std::size_t nCommonColumnsInCluster, RSealedPageMergeData &sealedPageData,
-                                          const RNTupleMergeData &mergeData, ROOT::Internal::RPageAllocator &pageAlloc);
+   ROOT::RResult<void>
+   MergeCommonColumns(ROOT::Internal::RClusterPool &clusterPool, const ROOT::RClusterDescriptor &clusterDesc,
+                      std::span<RColumnMergeInfo> commonColumns,
+                      const ROOT::Internal::RCluster::ColumnSet_t &commonColumnSet, std::size_t nCommonColumnsInCluster,
+                      RSealedPageMergeData &sealedPageData, const RNTupleMergeData &mergeData,
+                      ROOT::Internal::RPageAllocator &pageAlloc);
 
    [[nodiscard]]
    ROOT::RResult<void>
-   MergeSourceClusters(ROOT::Internal::RPageSource &source, std::span<const RColumnMergeInfo> commonColumns,
+   MergeSourceClusters(ROOT::Internal::RPageSource &source, std::span<RColumnMergeInfo> commonColumns,
                        std::span<const RColumnMergeInfo> extraDstColumns, RNTupleMergeData &mergeData);
 
    /// Creates a RNTupleMerger with the given destination.

tree/ntuple/inc/ROOT/RPageStorage.hxx

@@ -544,6 +544,15 @@ public:
    [[nodiscard]] std::unique_ptr<RNTupleModel>
    InitFromDescriptor(const ROOT::RNTupleDescriptor &descriptor, bool copyClusters);
 
+   /// Adds a new column representation to the given field.
+   /// \return The physical id of the first newly added column.
+   ROOT::DescriptorId_t
+   AddColumnRepresentation(const ROOT::RFieldDescriptor &field, std::span<const ENTupleColumnType> newRepresentation);
+
+   /// Adds a new alias column pointing to an existing column with the given physical id to the given field.
+   void AddAliasColumn(const ROOT::RNTupleDescriptor &desc, const ROOT::RFieldDescriptor &field,
+                       ROOT::DescriptorId_t physicalId);
+
    void CommitSuppressedColumn(ColumnHandle_t columnHandle) final;
    void CommitPage(ColumnHandle_t columnHandle, const ROOT::Internal::RPage &page) final;
    void CommitSealedPage(ROOT::DescriptorId_t physicalColumnId, const RPageStorage::RSealedPage &sealedPage) final;

tree/ntuple/src/RFieldBase.cxx

@@ -668,14 +668,14 @@ void ROOT::RFieldBase::Attach(std::unique_ptr<ROOT::RFieldBase> child, std::stri
 
 ROOT::NTupleSize_t ROOT::RFieldBase::EntryToColumnElementIndex(ROOT::NTupleSize_t globalIndex) const
 {
-   std::size_t result = globalIndex;
+   ROOT::NTupleSize_t result = globalIndex;
    for (auto f = this; f != nullptr; f = f->GetParent()) {
       auto parent = f->GetParent();
       if (parent && (parent->GetStructure() == ROOT::ENTupleStructure::kCollection ||
                      parent->GetStructure() == ROOT::ENTupleStructure::kVariant)) {
          return 0U;
       }
-      result *= std::max(f->GetNRepetitions(), std::size_t{1U});
+      result *= std::max<ROOT::NTupleSize_t>(f->GetNRepetitions(), ROOT::NTupleSize_t{1U});
    }
    return result;
 }

tree/ntuple/src/RNTupleDescriptor.cxx

@@ -961,8 +961,16 @@ ROOT::Internal::RClusterDescriptorBuilder::AddExtendedColumnRanges(const RNTuple
                // `ROOT::RFieldBase::EntryToColumnElementIndex()`, i.e. it is a principal column reachable from the
                // field zero excluding subfields of collection and variant fields.
                if (c.IsDeferredColumn()) {
-                  columnRange.SetFirstElementIndex(fCluster.GetFirstEntryIndex() * nRepetitions);
-                  columnRange.SetNElements(fCluster.GetNEntries() * nRepetitions);
+                  if (c.GetRepresentationIndex() == 0) {
+                     columnRange.SetFirstElementIndex(fCluster.GetFirstEntryIndex() * nRepetitions);
+                     columnRange.SetNElements(fCluster.GetNEntries() * nRepetitions);
+                  } else {
+                     const auto &field = desc.GetFieldDescriptor(fieldId);
+                     const auto firstReprColumnId = field.GetLogicalColumnIds()[c.GetIndex()];
+                     const auto &firstReprColumnRange = fCluster.fColumnRanges[firstReprColumnId];
+                     columnRange.SetFirstElementIndex(firstReprColumnRange.GetFirstElementIndex());
+                     columnRange.SetNElements(firstReprColumnRange.GetNElements());
+                  }
                   if (!columnRange.IsSuppressed()) {
                      auto &pageRange = fCluster.fPageRanges[physicalId];
                      pageRange.fPhysicalColumnId = physicalId;