Skip to content

[ntuple] Reject incompatible late class schemas #23684

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Elvand-Lie wants to merge 1 commit into
base: master
Choose a base branch
from
+259 −9
Open

[ntuple] Reject incompatible late class schemas #23684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions tree/ntuple/inc/ROOT/RNTupleMerger.hxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@ enum class ENTupleMergingMode {
kStrict,
/// The merger will update the output model to include all columns from all sources. Entries corresponding to columns
/// that are not present in a source will be set to the default value of the type.
/// When a field first appears in a later source, its class dictionary must match its on-disk schema and class
/// version. Detected schema mismatches or field creation errors are skipped or rejected according to fErrBehavior.
kUnion
};

Expand Down
94 changes: 85 additions & 9 deletions tree/ntuple/src/RNTupleMerger.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -736,11 +736,78 @@ static ROOT::RResult<void> RestoreRealFieldParams(ROOT::RFieldBase &field, const
return ROOT::RResult<void>::Success();
}

// Late extension connects reconstructed fields to a sink, so reader-side schema evolution cannot
// reconcile dictionary changes. Require the source layout before copying pages by field order.
[[nodiscard]]
static ROOT::RResult<void>
ValidateLateFieldSchema(const ROOT::RFieldBase &field, const ROOT::RFieldDescriptor &fieldDesc,
const ROOT::RNTupleDescriptor &srcDesc)
{
const auto &children = field.GetConstSubfields();
const auto &childIds = fieldDesc.GetLinkIds();
const auto checksum = fieldDesc.GetTypeChecksum();
std::string difference;
if (field.GetTypeName() != srcDesc.GetTypeNameForComparison(fieldDesc))
difference = "type name";
else if (field.GetStructure() != fieldDesc.GetStructure())
difference = "structure";
else if (field.GetNRepetitions() != fieldDesc.GetNRepetitions())
difference = "repetitions";
else if (field.GetFieldVersion() != fieldDesc.GetFieldVersion())
difference = "field version";
else if (field.GetTypeVersion() != fieldDesc.GetTypeVersion())
difference = "type version";
else if (checksum && *checksum != field.GetTypeChecksum())
difference = "type checksum";
else if (children.size() != childIds.size())
difference = "number of members";
if (!difference.empty()) {
return R__FAIL("cannot extend destination with field `" + srcDesc.GetQualifiedFieldName(fieldDesc.GetId()) +
"` of type `" + fieldDesc.GetTypeName() +
"`: its in-memory schema does not match the source schema (" + difference + ")");
}
for (std::size_t i = 0; i < children.size(); ++i) {
const auto &childDesc = srcDesc.GetFieldDescriptor(childIds[i]);
if (children[i]->GetFieldName() != childDesc.GetFieldName()) {
return R__FAIL("cannot extend destination with field `" + srcDesc.GetQualifiedFieldName(fieldDesc.GetId()) +
"` of type `" + fieldDesc.GetTypeName() +
"`: its in-memory member order does not match the source schema");
}
auto validated = ValidateLateFieldSchema(*children[i], childDesc, srcDesc);
if (!validated)
return R__FORWARD_ERROR(validated);
}
return ROOT::RResult<void>::Success();
}

[[nodiscard]]
static ROOT::RResult<std::vector<std::unique_ptr<ROOT::RFieldBase>>>
CreateLateFields(const std::vector<const ROOT::RFieldDescriptor *> &fieldDescs, const ROOT::RNTupleDescriptor &srcDesc)
{
std::vector<std::unique_ptr<ROOT::RFieldBase>> fields;
fields.reserve(fieldDescs.size());
for (const auto *fieldDesc : fieldDescs) {
std::unique_ptr<ROOT::RFieldBase> field;
try {
field = fieldDesc->CreateField(srcDesc);
} catch (const ROOT::RException &ex) {
return R__FAIL("cannot extend destination with field `" + srcDesc.GetQualifiedFieldName(fieldDesc->GetId()) +
"` of type `" + fieldDesc->GetTypeName() + "`: " + ex.what());
}
auto validated = ValidateLateFieldSchema(*field, *fieldDesc, srcDesc);
if (!validated)
return R__FORWARD_ERROR(validated);
fields.emplace_back(std::move(field));
}
return ROOT::RResult(std::move(fields));
}

// Applies late model extension to `mergeData.fDestination`, adding all `descCmp.fExtraSrcFields` to it.
[[nodiscard]]
static ROOT::RResult<void>
ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstModel,
ROOT::Internal::RPageSink &destination, const ROOT::RNTupleDescriptor &srcDesc, bool verbose)
ROOT::Internal::RPageSink &destination, const ROOT::RNTupleDescriptor &srcDesc, bool verbose,
std::vector<std::unique_ptr<ROOT::RFieldBase>> fields)
{
const auto &newFields = descCmp.fExtraSrcFields;
auto &commonFields = descCmp.fCommonFields;
Expand All @@ -762,11 +829,12 @@ ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstM

changeset.fAddedFields.reserve(newFields.size());
// First add all non-projected fields...
for (const auto *fieldDesc : newFields) {
for (std::size_t i = 0; i < newFields.size(); ++i) {
const auto *fieldDesc = newFields[i];
if (fieldDesc->IsProjectedField())
continue;

auto field = fieldDesc->CreateField(srcDesc);
auto field = std::move(fields[i]);
// Explicitly set the field representatives. This prevents UpdateSchema() from changing our column
// representations via AutoAdjustColumnTypes.
ROOT::RFieldBase::ColumnRepresentation_t representatives;
Expand All @@ -783,12 +851,13 @@ ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstM
changeset.AddField(std::move(field));
}
// ...then add all projected fields.
for (const auto *fieldDesc : newFields) {
for (std::size_t i = 0; i < newFields.size(); ++i) {
const auto *fieldDesc = newFields[i];
if (!fieldDesc->IsProjectedField())
continue;

ROOT::Internal::RProjectedFields::FieldMap_t fieldMap;
auto field = fieldDesc->CreateField(srcDesc);
auto field = std::move(fields[i]);
const auto sourceId = fieldDesc->GetProjectionSourceId();
const auto &sourceField = dstModel.GetConstField(srcDesc.GetQualifiedFieldName(sourceId));
fieldMap[field.get()] = &sourceField;
Expand Down Expand Up @@ -943,7 +1012,8 @@ class RNTupleMergeStrategy {

virtual void InitSource(RPageSource &source) = 0;
[[nodiscard]] virtual ROOT::RResult<RDescriptorsComparison> CompareSrcAndDstDescriptors() = 0;
[[nodiscard]] virtual ROOT::RResult<void> ExtendDestinationModel(RDescriptorsComparison &descCmp) = 0;
[[nodiscard]] virtual ROOT::RResult<void>
ExtendDestinationModel(RDescriptorsComparison &descCmp, std::vector<std::unique_ptr<ROOT::RFieldBase>> fields) = 0;
[[nodiscard]] virtual ROOT::RResult<void> MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) = 0;
virtual void BeforeCommit() = 0;
};
Expand Down Expand Up @@ -1010,10 +1080,12 @@ class RNTupleMergeStrategyDefault final : public RNTupleMergeStrategy {
return CompareDescriptorStructure(fDstDescriptor, *fSrcDescriptor);
}

[[nodiscard]] ROOT::RResult<void> ExtendDestinationModel(RDescriptorsComparison &descCmp) final
[[nodiscard]] ROOT::RResult<void>
ExtendDestinationModel(RDescriptorsComparison &descCmp, std::vector<std::unique_ptr<ROOT::RFieldBase>> fields) final
{
// late model extension for all fExtraSrcFields in Union mode
auto res = ::ExtendDestinationModel(descCmp, *fDstModel, fDestination, *fSrcDescriptor, fMergeOpts.fExtraVerbose);
auto res = ::ExtendDestinationModel(descCmp, *fDstModel, fDestination, *fSrcDescriptor, fMergeOpts.fExtraVerbose,
std::move(fields));
return res;
}

Expand Down Expand Up @@ -1606,7 +1678,11 @@ ROOT::RResult<void> RNTupleMerger::Merge(std::span<RPageSource *> sources, const
if (!descCmp.fExtraSrcFields.empty()) {
if (mergeOpts.fMergingMode == ENTupleMergingMode::kUnion) {
// late model extension for all fExtraSrcFields in Union mode
auto res = strat->ExtendDestinationModel(descCmp);
auto fields = CreateLateFields(descCmp.fExtraSrcFields, source->GetSharedDescriptorGuard().GetRef());
if (!fields) {
SKIP_OR_ABORT(fields.GetError()->GetReport())
}
auto res = strat->ExtendDestinationModel(descCmp, fields.Unwrap());
if (!res)
return R__FORWARD_ERROR(res);
} else if (mergeOpts.fMergingMode == ENTupleMergingMode::kStrict) {
Expand Down
172 changes: 172 additions & 0 deletions tree/ntuple/test/ntuple_evolution_shape.cxx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,181 @@

#include <TInterpreter.h>

#include <array>
#include <string>
#include <string_view>

TEST(RNTupleEvolution, UnionMergeLateClassSchema)
{
struct SchemaCase {
const char *fName;
const char *fMembers;
bool fMatches;
bool fNested;
bool fProjected = false;
bool fSkip = false;
bool fStreamer = false;
};
const std::array<SchemaCase, 16> cases{{
{"matching", "int fA = 11; int fC = 33;", true, false},
{"inserted", "int fA = 11; int fB = 22; int fC = 33;", false, false},
{"appended", "int fA = 11; int fC = 33; int fB = 22;", false, false},
{"removed", "int fA = 11;", false, false},
{"reordered", "int fC = 33; int fA = 11;", false, false},
{"type_changed", "float fA = 11; int fC = 33;", false, false},
{"nested_matching", "int fA = 11; int fC = 33;", true, true},
{"nested_inserted", "int fA = 11; int fB = 22; int fC = 33;", false, true},
{"projected_matching", "int fA = 11; int fC = 33;", true, false, true},
{"projected_inserted", "int fA = 11; int fB = 22; int fC = 33;", false, false, true},
{"skip_inserted", "int fA = 11; int fB = 22; int fC = 33;", false, false, false, true},
{"version_changed", "int fA = 11; int fC = 33; ClassDefNV(LateMergeClass, 2)", false, false},
{"streamer_common_matching", "int fA = 11; int fC = 33;", true, false, false, false, true},
{"streamer_evolved", "int fA = 11; int fB = 22; int fC = 33; ClassDefNV(LateMergeClass, 2)", false, false, false,
false, true},
{"missing_dictionary", nullptr, false, false},
{"skip_missing_dictionary", nullptr, false, false, false, true},
}};
for (const auto &testCase : cases) {
SCOPED_TRACE(testCase.fName);
// Each process gets its own dictionary; only the source writer sees the original definition.
ExecInFork(([&] {
FileRaii first("test_ntuple_merge_class_schema_first.root");
FileRaii second("test_ntuple_merge_class_schema_second.root");
FileRaii third("test_ntuple_merge_class_schema_third.root");
FileRaii output("test_ntuple_merge_class_schema_output.root");
const std::string typeName = testCase.fNested ? "std::vector<LateMergeClass>" : "LateMergeClass";
ExecInFork([&] {
first.PreserveFile();
second.PreserveFile();
ASSERT_TRUE(gInterpreter->Declare("struct LateMergeClass { int fA = 11; int fC = 33; };"));
{
auto model = RNTupleModel::Create();
model->MakeField<int>("e");
// Adding late streamer fields is unsupported; check matching streamer fields present from the start.
if (testCase.fStreamer && testCase.fMatches)
model->AddField(std::make_unique<ROOT::RStreamerField>("record", typeName));
else if (testCase.fProjected)
model->AddField(RFieldBase::Create("record", typeName).Unwrap());
auto writer = RNTupleWriter::Recreate(std::move(model), "ntpl", first.GetPath());
writer->Fill();
}
auto model = RNTupleModel::Create();
model->MakeField<int>("e");
*model->MakeField<int>("before_record") = 77;
if (testCase.fStreamer)
model->AddField(std::make_unique<ROOT::RStreamerField>("record", typeName));
else
model->AddField(RFieldBase::Create("record", typeName).Unwrap());
if (testCase.fProjected) {
model->AddProjectedField(RFieldBase::Create("alias", typeName).Unwrap(), [](const std::string &name) {
auto mapped = name;
mapped.replace(0, 5, "record");
return mapped;
});
}
auto writer = RNTupleWriter::Recreate(std::move(model), "ntpl", second.GetPath());
if (testCase.fNested) {
void *ptr = writer->GetModel().GetDefaultEntry().GetPtr<void>("record").get();
DeclarePointer(typeName, "values", ptr);
ProcessLine("values->emplace_back();");
}
writer->Fill();
});
if (testCase.fMembers) {
ASSERT_TRUE(
gInterpreter->Declare((std::string("struct LateMergeClass { ") + testCase.fMembers + " };").c_str()));
}
auto firstSource = RPageSource::Create("ntpl", first.GetPath());
auto secondSource = RPageSource::Create("ntpl", second.GetPath());
std::vector<RPageSource *> sources{firstSource.get(), secondSource.get()};
std::unique_ptr<RPageSource> thirdSource;
if (testCase.fSkip) {
{
auto model = RNTupleModel::Create();
model->MakeField<int>("e");
*model->MakeField<int>("before_record") = 99;
auto writer = RNTupleWriter::Recreate(std::move(model), "ntpl", third.GetPath());
writer->Fill();
}
thirdSource = RPageSource::Create("ntpl", third.GetPath());
sources.push_back(thirdSource.get());
}
auto destination = std::make_unique<RPageSinkFile>("ntpl", output.GetPath(), RNTupleWriteOptions());
auto *sink = destination.get();
RNTupleMerger merger{std::move(destination)};
RNTupleMergeOptions options;
options.fMergingMode = ENTupleMergingMode::kUnion;
if (testCase.fSkip)
options.fErrBehavior = ROOT::Experimental::Internal::ENTupleMergeErrBehavior::kSkip;
auto result = [&] {
ROOT::TestSupport::CheckDiagsRAII diags;
diags.optionalDiag(kWarning, "RIoUring", "io_uring is not available because:", false);
diags.optionalDiag(kWarning, "RRawFileUnix",
"io_uring setup failed, falling back to blocking I/O in ReadV");
if (!testCase.fMembers)
diags.optionalDiag(kWarning, "TClass::Init", "no dictionary for class LateMergeClass is available");
if (testCase.fSkip) {
diags.requiredDiag(kWarning, "ROOT.NTuple.Merge",
"Skipping RNTuple due to: cannot extend destination with field `record`", false);
}
return merger.Merge(sources, options);
}();
ASSERT_FALSE(::testing::Test::HasFailure());
if (testCase.fSkip) {
ASSERT_TRUE(bool(result)) << result.GetError()->GetReport();
auto reader = RNTupleReader::Open("ntpl", output.GetPath());
ASSERT_EQ(2u, reader->GetNEntries());
ASSERT_EQ(ROOT::kInvalidDescriptorId, reader->GetDescriptor().FindFieldId("record"));
auto before = reader->GetView<int>("before_record");
ASSERT_EQ(0, before(0));
ASSERT_EQ(99, before(1));
} else if (!testCase.fMatches) {
ASSERT_FALSE(bool(result));
ASSERT_NE(std::string::npos,
result.GetError()->GetReport().find(testCase.fMembers ? "does not match the source schema"
: "unknown type: LateMergeClass"));
ASSERT_EQ(ROOT::kInvalidDescriptorId,
sink->GetDescriptor().FindFieldId(testCase.fProjected ? "alias" : "record"));
ASSERT_EQ(ROOT::kInvalidDescriptorId, sink->GetDescriptor().FindFieldId("before_record"));
} else {
ASSERT_TRUE(bool(result)) << result.GetError()->GetReport();
auto reader = RNTupleReader::Open("ntpl", output.GetPath());
ASSERT_EQ(2u, reader->GetNEntries());
if (testCase.fStreamer) {
void *ptr = reader->GetModel().GetDefaultEntry().GetPtr<void>("record").get();
DeclarePointer(typeName, "mergedRecord", ptr);
reader->LoadEntry(1);
ASSERT_EQ(11, gInterpreter->Calc("mergedRecord->fA"));
ASSERT_EQ(33, gInterpreter->Calc("mergedRecord->fC"));
} else if (!testCase.fNested) {
auto a = reader->GetView<int>("record.fA");
auto c = reader->GetView<int>("record.fC");
ASSERT_EQ(testCase.fProjected ? 11 : 0, a(0));
ASSERT_EQ(testCase.fProjected ? 33 : 0, c(0));
ASSERT_EQ(11, a(1));
ASSERT_EQ(33, c(1));
if (testCase.fProjected) {
auto aliasA = reader->GetView<int>("alias.fA");
auto aliasC = reader->GetView<int>("alias.fC");
ASSERT_EQ(11, aliasA(0));
ASSERT_EQ(33, aliasC(0));
ASSERT_EQ(11, aliasA(1));
ASSERT_EQ(33, aliasC(1));
}
} else {
auto records = reader->GetCollectionView("record");
ASSERT_EQ(0u, records(0));
ASSERT_EQ(1u, records(1));
auto a = records.GetView<int>("_0.fA");
auto c = records.GetView<int>("_0.fC");
ASSERT_EQ(11, a(0));
ASSERT_EQ(33, c(0));
}
}
}));
}
}

TEST(RNTupleEvolution, AddedMember)
{
FileRaii fileGuard("test_ntuple_evolution_added_member.root");
Expand Down