[ntuple] add support for attribute merging

tree/ntuple/inc/ROOT/RNTupleMerger.hxx

@@ -140,6 +140,12 @@ public:
       return ENTupleMergeActionResult::kEmit;
    }
 
+   // Invoked at the end of each source's merging process.
+   virtual ENTupleMergeActionResult OnSourceEnd(ROOT::REntry &, const RNTupleMergeActionContext &)
+   {
+      return ENTupleMergeActionResult::kDrop;
+   }
+
    /// Invoked at the end of the merge process, right before committing the dataset.
    /// This allows appending more entries at the end of the merge.
    virtual ENTupleMergeActionResult BeforeCommit(ROOT::REntry &, const RNTupleMergeActionContext &)

tree/ntuple/src/RNTupleAttrWriting.cxx

@@ -17,12 +17,16 @@ static ROOT::RResult<void> ValidateAttributeModel(const ROOT::RNTupleModel &mode
 {
    const auto &projFields = ROOT::Internal::GetProjectedFieldsOfModel(model);
    if (!projFields.IsEmpty())
-      return R__FAIL("The Model used to create an AttributeSet cannot contain projected fields.");
+      return R__FAIL("The Model used to create an attribute set cannot contain projected fields.");
 
    for (const auto &field : model.GetConstFieldZero()) {
       if (field.GetStructure() == ROOT::ENTupleStructure::kStreamer)
-         return R__FAIL(std::string("The Model used to create an AttributeSet cannot contain Streamer field '") +
+         return R__FAIL(std::string("The Model used to create an attribute set cannot contain Streamer field '") +
                         field.GetQualifiedFieldName() + "'");
+      if (field.GetTypeName().empty())
+         return R__FAIL(
+            std::string("Untyped records or collections are currently unsupported as attribute types (field '") +
+            field.GetQualifiedFieldName() + "')");
    }
    return ROOT::RResult<void>::Success();
 }

tree/ntuple/test/CMakeLists.txt

@@ -43,6 +43,7 @@ if(NOT MSVC)
   # These unit tests rely on fork(), which is not available on Windows.
   ROOT_ADD_GTEST(ntuple_evolution_shape ntuple_evolution_shape.cxx LIBRARIES ROOTNTuple)
   ROOT_ADD_GTEST(ntuple_evolution_soa ntuple_evolution_soa.cxx LIBRARIES ROOTNTuple)
+  ROOT_ADD_GTEST(ntuple_evolution_merge ntuple_evolution_merge.cxx LIBRARIES ROOTNTuple)
   ROOT_ADD_GTEST(ntuple_emulated ntuple_emulated.cxx LIBRARIES ROOTNTuple)
   ROOT_ADD_GTEST(ntuple_merge_emulated ntuple_merge_emulated.cxx ntuple_test.cxx LIBRARIES ROOTNTuple xxHash::xxHash)
 endif()

tree/ntuple/test/ntuple_attributes.cxx

@@ -636,7 +636,7 @@ TEST(RNTupleAttributes, MultipleSetsTFile)
    auto reader = RNTupleReader::Open(*ntpl);
    EXPECT_EQ(reader->GetDescriptor().GetNAttributeSets(), 2);
 
-  // Extract the headers' offsets
+   // Extract the headers' offsets
    std::uint64_t mainHeaderOff = ntpl->GetSeekHeader();
    std::uint64_t attrSet1Off = 0, attrSet2Off = 0;
    int n = 0;
@@ -1348,3 +1348,24 @@ TEST(RNTupleAttributes, WriteZeroAttributes)
    auto iterCG = attrSetReader->GetDescriptor().GetClusterGroupIterable();
    EXPECT_EQ(iterCG.begin(), iterCG.end());
 }
+
+TEST(RNTupleAttributes, UntypedRecord)
+{
+   FileRaii fileGuard1("test_ntuple_attr_untyped_record_1.root");
+
+   auto model = RNTupleModel::Create();
+   auto fieldFoo = model->MakeField<int>("foo");
+   auto writer = RNTupleWriter::Recreate(std::move(model), "ntuple", fileGuard1.GetPath());
+   auto attrSetModel = RNTupleModel::Create();
+   std::vector<std::unique_ptr<RFieldBase>> children;
+   children.push_back(std::make_unique<RField<int>>("int"));
+   children.push_back(std::make_unique<RField<char>>("char"));
+   auto record = std::make_unique<ROOT::RRecordField>("record", std::move(children));
+   attrSetModel->AddField(std::move(record));
+   try {
+      auto attrSet = writer->CreateAttributeSet(std::move(attrSetModel), "AttrSet1");
+      FAIL() << "using Untyped records as attributes should fail";
+   } catch (const ROOT::RException &ex) {
+      EXPECT_THAT(ex.what(), testing::HasSubstr("Untyped records or collections are currently unsupported"));
+   }
+}

tree/ntuple/test/ntuple_evolution_merge.cxx

@@ -0,0 +1,119 @@
+#include "ntuple_test.hxx"
+
+#include "ntuple_fork.hxx"
+#include "ntuple_test_evolution.hxx"
+
+#include <TInterpreter.h>
+
+#include <string>
+#include <string_view>
+
+TEST(RNTupleMerger, MergeWithAttributesSchemaEvolve)
+{
+   ROOT::TestSupport::CheckDiagsRAII diagsRAII;
+   diagsRAII.requiredDiag(kWarning, "ROOT.NTuple", "RNTuple Attributes are experimental", false);
+
+   // Write two test ntuples to be merged.
+   // These files have 1 attribute set with the same name and they contain a custom class field.
+   // The second class field has a newer version that has been schema evolved from the first.
+   FileRaii fileGuard1("test_ntuple_merge_attr_schemaevolve_1.root");
+   ExecInFork([&] {
+      fileGuard1.PreserveFile();
+
+      ASSERT_TRUE(gInterpreter->Declare(R"(
+struct MyClass {
+   int fInt1 = 1;
+   int fInt3 = 3;
+};
+)"));
+
+      auto model = RNTupleModel::Create();
+      auto pInt = model->MakeField<int>("mainInt");
+      auto writer = RNTupleWriter::Recreate(std::move(model), "ntuple", fileGuard1.GetPath());
+      auto attrModel = RNTupleModel::Create();
+      attrModel->AddField(RFieldBase::Create("attr", "MyClass").Unwrap());
+      auto attrSet = writer->CreateAttributeSet(std::move(attrModel), "AttrSet1");
+      void *pAttr = attrSet->GetModel().GetDefaultEntry().GetPtr<void>("attr").get();
+      auto attrRange = attrSet->BeginRange();
+      DeclarePointer("MyClass", "pMyClass", pAttr);
+      ProcessLine("pMyClass->fInt1 = 11;");
+      ProcessLine("pMyClass->fInt3 = 33;");
+      *pInt = 100;
+      writer->Fill();
+      attrSet->CommitRange(std::move(attrRange));
+
+      // Reset / close the writer and flush the file.
+      writer.reset();
+   });
+
+   ASSERT_TRUE(gInterpreter->Declare(R"(
+struct MyClass {
+   int fInt1 = 1;
+   int fInt2 = 2;
+   int fInt3 = 3;
+};
+)"));
+
+   FileRaii fileGuard2("test_ntuple_merge_attr_schemaevolve_2.root");
+   {
+      auto model = RNTupleModel::Create();
+      auto pInt = model->MakeField<int>("mainInt");
+      auto writer = RNTupleWriter::Recreate(std::move(model), "ntuple", fileGuard2.GetPath());
+      auto attrModel = RNTupleModel::Create();
+      attrModel->AddField(RFieldBase::Create("attr", "MyClass").Unwrap());
+      auto attrSet = writer->CreateAttributeSet(std::move(attrModel), "AttrSet1");
+      void *pAttr = attrSet->GetModel().GetDefaultEntry().GetPtr<void>("attr").get();
+      auto attrRange = attrSet->BeginRange();
+      DeclarePointer("MyClass", "pMyClass", pAttr);
+      ProcessLine("pMyClass->fInt1 = 22;");
+      ProcessLine("pMyClass->fInt2 = 44;");
+      ProcessLine("pMyClass->fInt3 = 66;");
+      *pInt = 200;
+      writer->Fill();
+      attrSet->CommitRange(std::move(attrRange));
+   }
+
+   std::vector<std::unique_ptr<RPageSource>> sources;
+   sources.push_back(RPageSource::Create("ntuple", fileGuard1.GetPath(), RNTupleReadOptions()));
+   sources.push_back(RPageSource::Create("ntuple", fileGuard2.GetPath(), RNTupleReadOptions()));
+   std::vector<RPageSource *> sourcePtrs;
+   for (const auto &s : sources) {
+      sourcePtrs.push_back(s.get());
+   }
+
+   // Now merge the inputs
+   for (const auto mmode : {ENTupleMergingMode::kFilter, ENTupleMergingMode::kStrict, ENTupleMergingMode::kUnion}) {
+      const auto modeString = ToString(mmode);
+      SCOPED_TRACE(std::string("with merging mode = ") + modeString);
+      FileRaii fileGuardOut("test_ntuple_merge_attr_schemaevolve_out.root");
+      {
+         auto destination = std::make_unique<RPageSinkFile>("ntuple", fileGuardOut.GetPath(), RNTupleWriteOptions());
+         RNTupleMerger merger{std::move(destination)};
+         RNTupleMergeOptions opts;
+         opts.fMergingMode = mmode;
+         auto res = merger.Merge(sourcePtrs, opts);
+         ASSERT_TRUE(bool(res)) << res.GetError()->GetReport();
+      }
+      {
+         auto reader = ROOT::RNTupleReader::Open("ntuple", fileGuardOut.GetPath());
+         EXPECT_EQ(reader->GetNEntries(), 2);
+         EXPECT_EQ(reader->GetDescriptor().GetNAttributeSets(), 1);
+         {
+            auto attrSet1 = reader->OpenAttributeSet("AttrSet1");
+            EXPECT_EQ(attrSet1->GetNEntries(), 2);
+            const auto &attrEntry1 = attrSet1->GetModel().GetDefaultEntry();
+            auto pAttr = attrEntry1.GetPtr<void>("attr").get();
+            const auto ptrName = std::string("pMyClass_") + modeString;
+            DeclarePointer("MyClass", ptrName, pAttr);
+            for (auto idx : attrSet1->GetAttributes()) {
+               auto range1 = attrSet1->LoadEntry(idx);
+               EXPECT_EQ(range1.GetFirst(), idx);
+               EXPECT_EQ(range1.GetLast(), idx);
+               EXPECT_EVALUATE_EQ((ptrName + "->fInt1").c_str(), 11 * (idx + 1));
+               EXPECT_EVALUATE_EQ((ptrName + "->fInt2").c_str(), idx == 1 ? 44 : 2);
+               EXPECT_EVALUATE_EQ((ptrName + "->fInt3").c_str(), 33 * (idx + 1));
+            }
+         }
+      }
+   }
+}