[ntuple] Add RPageSinkFile::CloneWithDifferentName

Writing analogue to RPageSourceFile::OpenWithDifferentAnchor, it allows
to derive a new Sink from an existing one using the same underlying
directory but writing to a different RNTuple.

Author: silverweed

tree/ntuple/inc/ROOT/RMiniFile.hxx

@@ -261,6 +261,13 @@ public:
    RNTupleFileWriter &operator=(RNTupleFileWriter &&other) = delete;
    ~RNTupleFileWriter();
 
+   /// Creates a new RNTupleFileWriter with the same underlying TDirectory as this but writing to a different
+   /// RNTuple named `ntupleName`. This class is not thread safe, so one of the two writers may be writing to the file
+   /// at the same time.
+   /// This method is currently only supported for TFile-based Writers and will throw an exception if that's not the
+   /// case.
+   std::unique_ptr<RNTupleFileWriter> CloneWithDifferentName(std::string_view ntupleName) const;
+
    /// Seek a simple writer to offset. Note that previous data is not flushed immediately, but only by the next write
    /// (if necessary).
    void Seek(std::uint64_t offset);
@@ -286,6 +293,8 @@ public:
    void UpdateStreamerInfos(const ROOT::Internal::RNTupleSerializer::StreamerInfoMap_t &streamerInfos);
    /// Writes the RNTuple key to the file so that the header and footer keys can be found
    void Commit(int compression = RCompressionSetting::EDefaults::kUseGeneralPurpose);
+
+   std::string_view GetNTupleName() const { return fNTupleName; }
 };
 
 } // namespace Internal

tree/ntuple/inc/ROOT/RPageNullSink.hxx

@@ -106,6 +106,11 @@ public:
    void CommitStagedClusters(std::span<RStagedCluster>) final {}
    void CommitClusterGroup() final {}
    void CommitDatasetImpl() final {}
+
+   std::unique_ptr<RPageSink> CloneWithDifferentName(std::string_view, const RNTupleWriteOptions &) const final
+   {
+      throw ROOT::RException(R__FAIL("cannot clone null sink"));
+   }
 };
 
 } // namespace Internal

tree/ntuple/inc/ROOT/RPageSinkBuf.hxx

@@ -152,6 +152,9 @@ public:
    void CommitDatasetImpl() final;
 
    RPage ReservePage(ColumnHandle_t columnHandle, std::size_t nElements) final;
+
+   std::unique_ptr<RPageSink>
+   CloneWithDifferentName(std::string_view name, const RNTupleWriteOptions &opts) const final;
 }; // RPageSinkBuf
 
 } // namespace Internal

tree/ntuple/inc/ROOT/RPageStorage.hxx

@@ -387,6 +387,9 @@ public:
    /// CommitClusterGroup (or the beginning of writing).
    virtual void CommitClusterGroup() = 0;
 
+   virtual std::unique_ptr<RPageSink>
+   CloneWithDifferentName(std::string_view name, const RNTupleWriteOptions &opts) const = 0;
+
    /// The registered callback is executed at the beginning of CommitDataset();
    void RegisterOnCommitDatasetCallback(Callback_t callback) { fOnDatasetCommitCallbacks.emplace_back(callback); }
    /// Run the registered callbacks and finalize the current cluster and the entrire data set.

tree/ntuple/inc/ROOT/RPageStorageDaos.hxx

@@ -140,6 +140,9 @@ protected:
 public:
    RPageSinkDaos(std::string_view ntupleName, std::string_view uri, const ROOT::RNTupleWriteOptions &options);
    ~RPageSinkDaos() override;
+
+   std::unique_ptr<ROOT::Internal::RPageSink>
+   CloneWithDifferentName(std::string_view name, const ROOT::RNTupleWriteOptions &opts) const final;
 }; // class RPageSinkDaos
 
 // clang-format off

tree/ntuple/inc/ROOT/RPageStorageFile.hxx

@@ -73,6 +73,7 @@ private:
    ROOT::Internal::RNTupleSerializer::StreamerInfoMap_t fInfosOfClassFields;
 
    RPageSinkFile(std::string_view ntupleName, const ROOT::RNTupleWriteOptions &options);
+   RPageSinkFile(std::unique_ptr<ROOT::Internal::RNTupleFileWriter> writer, const ROOT::RNTupleWriteOptions &options);
 
    /// We pass bytesPacked so that TFile::ls() reports a reasonable value for the compression ratio of the corresponding
    /// key. It is not strictly necessary to write and read the sealed page.
@@ -109,6 +110,11 @@ public:
    ~RPageSinkFile() override;
 
    void UpdateSchema(const ROOT::Internal::RNTupleModelChangeset &changeset, ROOT::NTupleSize_t firstEntry) final;
+
+   /// Creates a new sink that uses the same underlying file/directory but writes to a different RNTuple with the
+   /// given `name`.
+   std::unique_ptr<RPageSink>
+   CloneWithDifferentName(std::string_view name, const ROOT::RNTupleWriteOptions &opts) const override;
 }; // class RPageSinkFile
 
 // clang-format off

tree/ntuple/src/RMiniFile.cxx

@@ -1232,7 +1232,7 @@ ROOT::Internal::RNTupleFileWriter::RNTupleFileWriter(std::string_view name, std:
    fStreamerInfoMap[infoRNTuple->GetNumber()] = infoRNTuple;
 }
 
-ROOT::Internal::RNTupleFileWriter::~RNTupleFileWriter() {}
+ROOT::Internal::RNTupleFileWriter::~RNTupleFileWriter() = default;
 
 std::unique_ptr<ROOT::Internal::RNTupleFileWriter>
 ROOT::Internal::RNTupleFileWriter::Recreate(std::string_view ntupleName, std::string_view path,
@@ -1317,6 +1317,16 @@ ROOT::Internal::RNTupleFileWriter::Append(std::string_view ntupleName, ROOT::Exp
    return writer;
 }
 
+std::unique_ptr<ROOT::Internal::RNTupleFileWriter>
+ROOT::Internal::RNTupleFileWriter::CloneWithDifferentName(std::string_view ntupleName) const
+{
+   if (auto *file = std::get_if<RFileProper>(&fFile)) {
+      return Append(ntupleName, *file->fDirectory, fNTupleAnchor.fMaxKeySize);
+   }
+   // TODO: support also non-TFile-based writers
+   throw ROOT::RException(R__FAIL("cannot clone a non-TFile-based RNTupleFileWriter."));
+}
+
 void ROOT::Internal::RNTupleFileWriter::Seek(std::uint64_t offset)
 {
    RFileSimple *fileSimple = std::get_if<RFileSimple>(&fFile);

tree/ntuple/src/RNTupleParallelWriter.cxx

@@ -113,6 +113,11 @@ class RPageSynchronizingSink : public RPageSink {
    }
 
    RSinkGuard GetSinkGuard() final { return RSinkGuard(fMutex); }
+
+   std::unique_ptr<RPageSink> CloneWithDifferentName(std::string_view, const ROOT::RNTupleWriteOptions &) const final
+   {
+      throw ROOT::RException(R__FAIL("cloning a RPageSynchronizingSink is not implemented yet"));
+   }
 };
 
 } // namespace

tree/ntuple/src/RPageSinkBuf.cxx

@@ -321,3 +321,9 @@ ROOT::Internal::RPage ROOT::Internal::RPageSinkBuf::ReservePage(ColumnHandle_t c
 {
    return fInnerSink->ReservePage(columnHandle, nElements);
 }
+
+std::unique_ptr<ROOT::Internal::RPageSink>
+ROOT::Internal::RPageSinkBuf::CloneWithDifferentName(std::string_view name, const RNTupleWriteOptions &opts) const
+{
+   return fInnerSink->CloneWithDifferentName(name, opts);
+}

tree/ntuple/src/RPageStorageDaos.cxx

@@ -482,6 +482,13 @@ void ROOT::Experimental::Internal::RPageSinkDaos::WriteNTupleAnchor()
       kDistributionKeyDefault, kAttributeKeyAnchor, kCidMetadata);
 }
 
+std::unique_ptr<ROOT::Internal::RPageSink>
+ROOT::Experimental::Internal::RPageSinkDaos::CloneWithDifferentName(std::string_view /*name*/,
+                                                                    const ROOT::RNTupleWriteOptions & /*opts*/) const
+{
+   throw ROOT::RException(R__FAIL("cloning a DAOS sink is not implemented yet"));
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 ROOT::Experimental::Internal::RPageSourceDaos::RPageSourceDaos(std::string_view ntupleName, std::string_view uri,