[ntuple] Add RPagePersistentSink::AddColumnRepresentation

Internal functionality to be used by the Merger

Author: silverweed

tree/ntuple/inc/ROOT/RPageStorage.hxx

@@ -544,6 +544,9 @@ public:
    [[nodiscard]] std::unique_ptr<RNTupleModel>
    InitFromDescriptor(const ROOT::RNTupleDescriptor &descriptor, bool copyClusters);
 
+   void
+   AddColumnRepresentation(const ROOT::RFieldDescriptor &field, std::span<const ENTupleColumnType> newRepresentation);
+
    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/RPageStorage.cxx

@@ -1049,6 +1049,57 @@ ROOT::Internal::RPagePersistentSink::InitFromDescriptor(const ROOT::RNTupleDescr
    return model;
 }
 
+void ROOT::Internal::RPagePersistentSink::AddColumnRepresentation(const ROOT::RFieldDescriptor &field,
+                                                                  std::span<const ENTupleColumnType> newRepresentation)
+{
+   R__ASSERT(field.GetColumnCardinality() > 0);
+   R__ASSERT(!field.GetLogicalColumnIds().empty());
+   R__ASSERT(newRepresentation.size() == field.GetColumnCardinality());
+
+   const auto firstPhysicalIndex = fDescriptorBuilder.GetDescriptor().GetNPhysicalColumns();
+   const auto reprIndex = field.GetLogicalColumnIds().size() / field.GetColumnCardinality();
+
+   fDescriptorBuilder.ShiftAliasColumns(newRepresentation.size());
+
+   std::uint16_t columnIndex = 0; // index into the representation
+   for (auto columnType : newRepresentation) {
+      // Extending columns with variable bit width is currently unsupported.
+      const auto [rangeMin, rangeMax] = ROOT::Internal::RColumnElementBase::GetValidBitRange(columnType);
+      R__ASSERT(rangeMin == rangeMax);
+
+      const auto firstReprColumnId = field.GetLogicalColumnIds()[columnIndex];
+      const auto &firstReprColumnRange = fOpenColumnRanges.at(firstReprColumnId);
+      const auto columnId = firstPhysicalIndex + columnIndex;
+
+      RColumnDescriptorBuilder columnBuilder;
+      columnBuilder.LogicalColumnId(columnId)
+         .PhysicalColumnId(columnId)
+         .FieldId(field.GetId())
+         .BitsOnStorage(rangeMax)
+         .Type(columnType)
+         .Index(columnIndex)
+         // NOTE: marking this column as suppressed with the minus sign
+         .FirstElementIndex(-firstReprColumnRange.GetFirstElementIndex())
+         .RepresentationIndex(reprIndex);
+      fDescriptorBuilder.AddColumn(columnBuilder.MakeDescriptor().Unwrap());
+
+      ROOT::RClusterDescriptor::RColumnRange columnRange;
+      columnRange.SetPhysicalColumnId(columnId);
+      columnRange.SetFirstElementIndex(firstReprColumnRange.GetFirstElementIndex());
+      columnRange.SetNElements(0);
+      columnRange.SetCompressionSettings(GetWriteOptions().GetCompression());
+      fOpenColumnRanges.emplace_back(columnRange);
+
+      ROOT::RClusterDescriptor::RPageRange pageRange;
+      pageRange.SetPhysicalColumnId(columnId);
+      fOpenPageRanges.emplace_back(std::move(pageRange));
+
+      fSerializationContext.MapPhysicalColumnId(columnId);
+
+      ++columnIndex;
+   }
+}
+
 void ROOT::Internal::RPagePersistentSink::CommitSuppressedColumn(ColumnHandle_t columnHandle)
 {
    fOpenColumnRanges.at(columnHandle.fPhysicalId).SetIsSuppressed(true);