[ntuple] Add RPageSinkFile::CloneWithDifferentName

tree/ntuple/inc/ROOT/RMiniFile.hxx

@@ -261,6 +261,12 @@ 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`. Onle one of the two writers can safely write 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 +292,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<RImplTFile>(&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)
 {
    RImplSimple *fileSimple = std::get_if<RImplSimple>(&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,

tree/ntuple/src/RPageStorageFile.cxx

@@ -72,6 +72,13 @@ ROOT::Internal::RPageSinkFile::RPageSinkFile(std::string_view ntupleName, ROOT::
    fWriter = RNTupleFileWriter::Append(ntupleName, file, ntupleDir, options.GetMaxKeySize());
 }
 
+ROOT::Internal::RPageSinkFile::RPageSinkFile(std::unique_ptr<ROOT::Internal::RNTupleFileWriter> writer,
+                                             const ROOT::RNTupleWriteOptions &options)
+   : RPageSinkFile(writer->GetNTupleName(), options)
+{
+   fWriter = std::move(writer);
+}
+
 ROOT::Internal::RPageSinkFile::~RPageSinkFile() {}
 
 void ROOT::Internal::RPageSinkFile::InitImpl(unsigned char *serializedHeader, std::uint32_t length)
@@ -291,6 +298,15 @@ void ROOT::Internal::RPageSinkFile::CommitDatasetImpl(unsigned char *serializedF
    fWriter->Commit(GetWriteOptions().GetCompression());
 }
 
+std::unique_ptr<ROOT::Internal::RPageSink>
+ROOT::Internal::RPageSinkFile::CloneWithDifferentName(std::string_view name,
+                                                      const ROOT::RNTupleWriteOptions &opts) const
+{
+   auto writer = fWriter->CloneWithDifferentName(name);
+   auto cloned = std::unique_ptr<RPageSinkFile>(new RPageSinkFile(std::move(writer), opts));
+   return cloned;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 ROOT::Internal::RPageSourceFile::RPageSourceFile(std::string_view ntupleName, const ROOT::RNTupleReadOptions &opts)

tree/ntuple/test/ntuple_endian.cxx

@@ -60,6 +60,11 @@ class RPageSinkMock : public RPageSink {
    void CommitClusterGroup() final {}
    void CommitDatasetImpl() final {}
 
+   std::unique_ptr<RPageSink> CloneWithDifferentName(std::string_view, const ROOT::RNTupleWriteOptions &) const final
+   {
+      throw ROOT::RException(R__FAIL("cannot clone sink"));
+   }
+
 public:
    RPageSinkMock(const RColumnElementBase &elt) : RPageSink("test", ROOT::RNTupleWriteOptions()), fElement(elt)
    {

tree/ntuple/test/ntuple_storage.cxx

@@ -55,6 +55,11 @@ class RPageSinkMock : public RPageSink {
    void CommitClusterGroup() final {}
    void CommitDatasetImpl() final {}
 
+   std::unique_ptr<RPageSink> CloneWithDifferentName(std::string_view, const ROOT::RNTupleWriteOptions &) const final
+   {
+      throw ROOT::RException(R__FAIL("cannot clone sink"));
+   }
+
 public:
    RPageSinkMock(const ROOT::RNTupleWriteOptions &options) : RPageSink("test", options) {}
 };
@@ -1204,3 +1209,26 @@ TEST(RPageSourceFile, OpenDifferentAnchor)
       EXPECT_NE(desc2->FindFieldId("c"), ROOT::kInvalidDescriptorId);
    }
 }
+
+TEST(RPageSinkFile, CloneWithDifferentName)
+{
+   FileRaii fileGuard("test_ntuple_page_sink_file_diffname.ntuple");
+
+   auto file = std::unique_ptr<TFile>(TFile::Open(fileGuard.GetPath().c_str(), "RECREATE"));
+   auto opts = RNTupleWriteOptions();
+   auto model = RNTupleModel::Create();
+
+   {
+      auto sink1 = std::make_unique<RPageSinkFile>("ntuple1", *file, opts);
+      sink1->Init(*model);
+
+      auto sink2 = sink1->CloneWithDifferentName("ntuple2", opts);
+      sink2->Init(*model);
+
+      sink1->CommitDataset();
+      sink2->CommitDataset();
+   }
+
+   EXPECT_NE(file->Get<ROOT::RNTuple>("ntuple1"), nullptr);
+   EXPECT_NE(file->Get<ROOT::RNTuple>("ntuple2"), nullptr);
+}