From d72f1154404730d2a3c6e70420df8538446e5bb2 Mon Sep 17 00:00:00 2001 From: silverweed Date: Mon, 5 Oct 2026 09:08:30 +0200 Subject: [PATCH 1/5] [ntuple] Consistently re-create the sink in merger tests Otherwise it'll gonna reuse the already-initialized source, which works but it doesn't test the intended code path --- tree/ntuple/test/ntuple_merger.cxx | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tree/ntuple/test/ntuple_merger.cxx b/tree/ntuple/test/ntuple_merger.cxx index 27287d804d165..3c2d871b2d6fb 100644 --- a/tree/ntuple/test/ntuple_merger.cxx +++ b/tree/ntuple/test/ntuple_merger.cxx @@ -580,11 +580,10 @@ TEST(RNTupleMerger, MergeInconsistentTypes) } // Create the output - auto destination = std::make_unique("ntuple", fileGuard3.GetPath(), RNTupleWriteOptions()); - RNTupleMerger merger{std::move(destination)}; - // We expect this to fail since the fields between the sources do NOT match for (const auto mmode : {ENTupleMergingMode::kFilter, ENTupleMergingMode::kStrict, ENTupleMergingMode::kUnion}) { + auto destination = std::make_unique("ntuple", fileGuard3.GetPath(), RNTupleWriteOptions()); + RNTupleMerger merger{std::move(destination)}; RNTupleMergeOptions opts; opts.fMergingMode = mmode; auto res = merger.Merge(sourcePtrs, opts); @@ -4588,12 +4587,11 @@ TEST(RNTupleMerger, MergeNewerVersion) } // Create the output - auto destination = std::make_unique("ntuple", fileGuardOut.GetPath(), RNTupleWriteOptions()); - RNTupleMerger merger{std::move(destination)}; - for (const auto mmode : {ENTupleMergingMode::kFilter, ENTupleMergingMode::kStrict, ENTupleMergingMode::kUnion}) { CheckDiagsRAII diagsRaii; diagsRaii.requiredDiag(kWarning, "ROOT.NTuple.Merge", "has a higher format version", false); + auto destination = std::make_unique("ntuple", fileGuardOut.GetPath(), RNTupleWriteOptions()); + RNTupleMerger merger{std::move(destination)}; RNTupleMergeOptions opts; opts.fMergingMode = mmode; auto res = merger.Merge(sourcePtrs, opts); @@ -4602,6 +4600,8 @@ TEST(RNTupleMerger, MergeNewerVersion) // Now merge again but with VersionBehavior set to Abort. { + auto destination = std::make_unique("ntuple", fileGuardOut.GetPath(), RNTupleWriteOptions()); + RNTupleMerger merger{std::move(destination)}; RNTupleMergeOptions opts; opts.fVersionBehavior = ROOT::Experimental::Internal::ENTupleMergeVersionBehavior::kAbortOnHigherVersion; auto res = merger.Merge(sourcePtrs, opts); From c8dd1a00f9b12cbf6dcf6f99e9df889445bcbe90 Mon Sep 17 00:00:00 2001 From: silverweed Date: Tue, 20 Oct 2026 16:19:40 +0200 Subject: [PATCH 2/5] [ntuple] add CallSetArtificialOnField() For future use by the Merger --- tree/ntuple/inc/ROOT/RFieldBase.hxx | 2 ++ tree/ntuple/src/RFieldBase.cxx | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/tree/ntuple/inc/ROOT/RFieldBase.hxx b/tree/ntuple/inc/ROOT/RFieldBase.hxx index 70880e3e3cfda..037dd318edf0d 100644 --- a/tree/ntuple/inc/ROOT/RFieldBase.hxx +++ b/tree/ntuple/inc/ROOT/RFieldBase.hxx @@ -64,6 +64,7 @@ void CallConnectPageSourceOnField(RFieldBase &, ROOT::Internal::RPageSource &); ROOT::RResult> CallFieldBaseCreate(const std::string &fieldName, const std::string &typeName, const ROOT::RCreateFieldOptions &options, const ROOT::RNTupleDescriptor *desc, ROOT::DescriptorId_t fieldId); +void CallSetArtificialOnField(RFieldBase &); } // namespace Internal @@ -100,6 +101,7 @@ class RFieldBase { Internal::CallFieldBaseCreate(const std::string &fieldName, const std::string &typeName, const ROOT::RCreateFieldOptions &options, const ROOT::RNTupleDescriptor *desc, ROOT::DescriptorId_t fieldId); + friend void Internal::CallSetArtificialOnField(RFieldBase &); using ReadCallback_t = std::function; diff --git a/tree/ntuple/src/RFieldBase.cxx b/tree/ntuple/src/RFieldBase.cxx index 1de0bb530051f..848837e4a6252 100644 --- a/tree/ntuple/src/RFieldBase.cxx +++ b/tree/ntuple/src/RFieldBase.cxx @@ -85,6 +85,10 @@ void ROOT::Internal::CallConnectPageSourceOnField(RFieldBase &field, ROOT::Inter { field.ConnectPageSource(source); } +void ROOT::Internal::CallSetArtificialOnField(RFieldBase &field) +{ + field.SetArtificial(); +} ROOT::RResult> ROOT::Internal::CallFieldBaseCreate(const std::string &fieldName, const std::string &typeName, From 0fa29aca7e6a4ff7d2048eab7eda2a711e385af4 Mon Sep 17 00:00:00 2001 From: silverweed Date: Thu, 1 Oct 2026 10:49:18 +0200 Subject: [PATCH 3/5] [ntuple] Introduce slow merging strategy --- tree/ntuple/inc/ROOT/REntry.hxx | 2 + tree/ntuple/inc/ROOT/RNTupleMerger.hxx | 53 +++++ tree/ntuple/src/RNTupleMerger.cxx | 270 ++++++++++++++++++++++--- 3 files changed, 296 insertions(+), 29 deletions(-) diff --git a/tree/ntuple/inc/ROOT/REntry.hxx b/tree/ntuple/inc/ROOT/REntry.hxx index 590e39853eff2..4950c195c39f4 100644 --- a/tree/ntuple/inc/ROOT/REntry.hxx +++ b/tree/ntuple/inc/ROOT/REntry.hxx @@ -38,6 +38,7 @@ class RNTupleAttrSetReader; namespace Internal { struct RNTupleAttrEntry; +class RNTupleMergeStrategySlow; } } // namespace Experimental @@ -56,6 +57,7 @@ class REntry { friend class RNTupleModel; friend class RNTupleReader; friend class Experimental::RNTupleAttrSetReader; + friend class Experimental::Internal::RNTupleMergeStrategySlow; friend struct Experimental::Internal::RNTupleAttrEntry; private: diff --git a/tree/ntuple/inc/ROOT/RNTupleMerger.hxx b/tree/ntuple/inc/ROOT/RNTupleMerger.hxx index a50943da504cd..7dee28137a618 100644 --- a/tree/ntuple/inc/ROOT/RNTupleMerger.hxx +++ b/tree/ntuple/inc/ROOT/RNTupleMerger.hxx @@ -27,6 +27,7 @@ namespace ROOT { +class REntry; class RNTuple; namespace Internal { @@ -35,6 +36,10 @@ class RPageSource; namespace Experimental::Internal { +class RNTupleMergeStrategySlow; +struct RNTupleSlowMergeData; +struct RDescriptorsComparison; + enum class ENTupleMergingMode { /// The merger will discard all columns that aren't present in the prototype model (i.e. the model of the first /// source); also all subsequent RNTuples must contain at least all the columns that are present in the prototype @@ -95,6 +100,54 @@ struct RNTupleMergeOptions { bool fExtraVerbose = false; }; +/// Information regarding the current slow merge process that gets passed to the RNTupleMergeActions +struct RNTupleMergeActionContext { + ROOT::NTupleSize_t fSourceEntryIndex = 0; + ROOT::NTupleSize_t fDestinationEntryIndex = 0; +}; + +enum class ENTupleMergeActionResult { + // Don't emit this entry + kDrop, + // Emit this entry, then continue + kEmit, + // Emit this entry, then call the function again + kEmitMulti, +}; + +// clang-format off +/** + * \class ROOT::Experimental::Internal::RNTupleMergeActions + * \ingroup NTuple + * \brief Container for callbacks that are invoked during the "slow merge" process. + * This allows customizing how the merging happens entry-by-entry, enabling things like dropping individual + * entries from the destination RNTuple, modifying their values on the fly or creating new entries artificially. + */ +// clang-format on +class RNTupleMergeActions { +public: + RNTupleMergeActions() = default; + virtual ~RNTupleMergeActions() = default; + + RNTupleMergeActions(const RNTupleMergeActions &) = delete; + RNTupleMergeActions &operator=(const RNTupleMergeActions &) = delete; + RNTupleMergeActions(RNTupleMergeActions &&) = default; + RNTupleMergeActions &operator=(RNTupleMergeActions &&) = default; + + /// Invoked every time an entry is merged from a source RNTuple to the destination RNTuple. + virtual ENTupleMergeActionResult OnFill(ROOT::REntry &, const RNTupleMergeActionContext &) + { + return ENTupleMergeActionResult::kEmit; + } + + /// 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 &) + { + return ENTupleMergeActionResult::kDrop; + } +}; + // clang-format off /** * \class ROOT::Experimental::Internal::RNTupleMerger diff --git a/tree/ntuple/src/RNTupleMerger.cxx b/tree/ntuple/src/RNTupleMerger.cxx index 95cae963295ea..c145cf08c4e8d 100644 --- a/tree/ntuple/src/RNTupleMerger.cxx +++ b/tree/ntuple/src/RNTupleMerger.cxx @@ -269,6 +269,11 @@ try { return -1; } +// ========================================================================================= +// +// Local types and functions +// +// ========================================================================================= namespace { // Functor used to change the compression of a page to `fCompressionSettings`. struct RChangeCompressionFunc { @@ -366,6 +371,9 @@ struct RColumnOutInfo { ROOT::DescriptorId_t fColumnId = ROOT::kInvalidDescriptorId; }; +// { ".fully.qualified.fieldName.colInputIndex.colOutputReprIndex" => colOutputInfo } +using ColumnIdMap_t = std::unordered_map; + struct RColumnMergeInfo { // This column name is built as a dot-separated concatenation of the ancestry of // the columns' parent fields' names plus the index of the column itself. @@ -382,14 +390,6 @@ struct RColumnMergeInfo { const ROOT::RNTupleDescriptor *fParentNTupleDescriptor = nullptr; }; -// { ".fully.qualified.fieldName.colInputIndex.colOutputReprIndex" => colOutputInfo } -using ColumnIdMap_t = std::unordered_map; - -struct RColumnInfoGroup { - std::vector fExtraDstColumns; - std::vector fCommonColumns; -}; - struct RSealedPageMergeData { // We use a std::deque so that references to the contained SealedPageSequence_t, and its iterators, are // never invalidated. @@ -398,6 +398,15 @@ struct RSealedPageMergeData { std::vector> fBuffers; }; +struct RColumnInfoGroup { + std::vector fExtraDstColumns; + std::vector fCommonColumns; +}; + +} // namespace + +// These structs cannot be in the anon namespace becase they're used in RNTupleMerger's private interface. +namespace ROOT::Experimental::Internal { template using FieldCollectionMap_t = std::unordered_map>; @@ -411,8 +420,7 @@ struct RDescriptorsComparison { FieldCollectionMap_t fColReprMappings; FieldCollectionMap_t fColReprExtensions; }; - -} // namespace +} // namespace ROOT::Experimental::Internal // Subprocedure of CompareDescriptorStructure, extracted for readability. // Given two fields, attempts to match their column representations and schedules column extensions if necessary. @@ -739,7 +747,8 @@ static ROOT::RResult RestoreRealFieldParams(ROOT::RFieldBase &field, const [[nodiscard]] static ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstModel, - ROOT::Internal::RPageSink &destination, const ROOT::RNTupleDescriptor &srcDesc, bool verbose) + ROOT::Internal::RPageSink &destination, const ROOT::RNTupleDescriptor &srcDesc, + ROOT::NTupleSize_t nEntries, bool verbose, bool forSlowMerge) { const auto &newFields = descCmp.fExtraSrcFields; auto &commonFields = descCmp.fCommonFields; @@ -768,19 +777,21 @@ ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstM ROOT::RCreateFieldOptions cfOpts; cfOpts.SetEmulateUnknownTypes(true); auto field = fieldDesc->CreateField(srcDesc, cfOpts); - // Explicitly set the field representatives. This prevents UpdateSchema() from changing our column - // representations via AutoAdjustColumnTypes. - ROOT::RFieldBase::ColumnRepresentation_t representatives; - for (const auto &colId : fieldDesc->GetLogicalColumnIds()) { - const auto &column = srcDesc.GetColumnDescriptor(colId); - representatives.push_back(column.GetType()); + if (!forSlowMerge) { + // Explicitly set the field representatives. This prevents UpdateSchema() from changing our column + // representations via AutoAdjustColumnTypes. + ROOT::RFieldBase::ColumnRepresentation_t representatives; + for (const auto &colId : fieldDesc->GetLogicalColumnIds()) { + const auto &column = srcDesc.GetColumnDescriptor(colId); + representatives.push_back(column.GetType()); + } + field->SetColumnRepresentatives({representatives}); + // Pinning the representative is not enough for truncated/quantized columns: their parameters + // are stored in the in-memory field, which CreateField() leaves at defaults. + auto restored = RestoreRealFieldParams(*field, *fieldDesc, srcDesc); + if (!restored) + return R__FORWARD_ERROR(restored); } - field->SetColumnRepresentatives({representatives}); - // Pinning the representative is not enough for truncated/quantized columns: their parameters - // are stored in the in-memory field, which CreateField() leaves at defaults. - auto restored = RestoreRealFieldParams(*field, *fieldDesc, srcDesc); - if (!restored) - return R__FORWARD_ERROR(restored); changeset.AddField(std::move(field)); } // ...then add all projected fields. @@ -812,7 +823,7 @@ ExtendDestinationModel(RDescriptorsComparison &descCmp, ROOT::RNTupleModel &dstM // need to connect these either in principle). // NOTE: this calls AutoAdjustColumnTypes, but we have set the column representations of all fields // explicitly, so it will not change it under the hood. - destination.UpdateSchema(changeset, destination.GetNEntries()); + destination.UpdateSchema(changeset, nEntries); } catch (const ROOT::RException &ex) { return R__FAIL(ex.what()); } @@ -924,15 +935,16 @@ static void AddColumnExtensionsInFieldOrder( } namespace { +// Collection of common methods and data to be used for any kind of RNTuples merging. class RNTupleMergeStrategy { protected: - // Fixed during the whole merging + // Fixed during the whole merging. Note that these references are owned by the RNTupleMerger and share its lifetime. std::unique_ptr &fDstModel; ROOT::Internal::RPagePersistentSink &fDestination; const ROOT::RNTupleDescriptor &fDstDescriptor; RNTupleMergeOptions fMergeOpts; - // Changes at every source + // Changes at every source. InitSource() is supposed to update this right after Attaching the source. const ROOT::RNTupleDescriptor *fSrcDescriptor = nullptr; public: @@ -944,13 +956,30 @@ class RNTupleMergeStrategy { virtual ~RNTupleMergeStrategy() = default; + // Invoked as the first thing when we start processing a new source virtual void InitSource(RPageSource &source) = 0; + + // Invoked before merging starts, right after initting the source. Determines whether the current source's schema is + // compatible with the destination's. [[nodiscard]] virtual ROOT::RResult CompareSrcAndDstDescriptors() = 0; + + // Invoked when the destination model must be late model extended (only invoked in case of union merging) [[nodiscard]] virtual ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) = 0; + + // Invoked when merging one source [[nodiscard]] virtual ROOT::RResult MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) = 0; + + // Invoked at the end of the whole process, just before committing the dataset virtual void BeforeCommit() = 0; }; +// ========================================================================================= +// +// MergeStrategyDefault +// +// ========================================================================================= +// Strategy used for L1, L2 and L3 merging +// (https://github.com/root-project/root/blob/master/tree/ntuple/doc/Merging.md#glossary). class RNTupleMergeStrategyDefault final : public RNTupleMergeStrategy { std::unique_ptr fPageAlloc = std::make_unique(); std::optional fTaskGroup; @@ -1017,7 +1046,9 @@ class RNTupleMergeStrategyDefault final : public RNTupleMergeStrategy { [[nodiscard]] ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) 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, + fDestination.GetNEntries(), fMergeOpts.fExtraVerbose, + /*forSlowMerge=*/false); return res; } @@ -1025,7 +1056,6 @@ class RNTupleMergeStrategyDefault final : public RNTupleMergeStrategy { [[nodiscard]] ROOT::RResult MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) final; }; - } // namespace // Given a field, fill `columns` and `mergeData.fColumnIdMap` with information about all columns belonging to it and @@ -1516,6 +1546,179 @@ ROOT::RResult RNTupleMergeStrategyDefault::MergeSource(RPageSource &source return res; } +// ========================================================================================= +// +// MergeStrategySlow +// +// ========================================================================================= +namespace ROOT::Experimental::Internal { +// Strategy used for L4 merging (https://github.com/root-project/root/blob/master/tree/ntuple/doc/Merging.md#glossary). +// This class must be in ROOT::Experimental::Internal because it befriends ROOT::REntry +class RNTupleMergeStrategySlow final : public RNTupleMergeStrategy { + std::unique_ptr fDstEntry; + std::unique_ptr fActions = std::make_unique(); + RNTupleMergeActionContext fMergeContext; + +public: + RNTupleMergeStrategySlow(std::unique_ptr &dstModel, ROOT::Internal::RPagePersistentSink &dst, + const RNTupleMergeOptions &mergeOpts) + : RNTupleMergeStrategy(dstModel, dst, mergeOpts) + { + // dstModel is non null iff we are doing an incremental merge. Currently we don't support it. + assert(!dstModel); + } + + void InitSource(RPageSource &source) final + { + source.Attach(); + fSrcDescriptor = &source.GetSharedDescriptorGuard().GetConstRef(); + + if (!fDstModel) { + ROOT::RNTupleDescriptor::RCreateModelOptions cmOpts; + cmOpts.SetCreateBare(true); + fDstModel = source.GetSharedDescriptorGuard()->CreateModel(cmOpts); + fDestination.Init(*fDstModel); + fDstEntry = fDstModel->CreateEntry(); + } + } + + ROOT::RResult CompareSrcAndDstDescriptors() final + { + return FillCommonAndExtraFields(fDstDescriptor, *fSrcDescriptor); + } + + [[nodiscard]] ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) final + { + // late model extension for all fExtraSrcFields in Union mode + auto res = ::ExtendDestinationModel(descCmp, *fDstModel, fDestination, *fSrcDescriptor, + fMergeContext.fDestinationEntryIndex, fMergeOpts.fExtraVerbose, + /*forSlowMerge=*/true); + // recreate the entry since the model has changed + fDstEntry = fDstModel->CreateEntry(); + return res; + } + + [[nodiscard]] ROOT::RResult MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) final; + + void BeforeCommit() final + { + while (true) { + auto res = fActions->BeforeCommit(*fDstEntry, fMergeContext); + if (res == ENTupleMergeActionResult::kDrop) + break; + fMergeContext.fDestinationEntryIndex += 1; + fDstEntry->Append(); + if (res == ENTupleMergeActionResult::kEmit) + break; + } + + // TODO: this should happen more often than once at the end and it should depend on the target cluster size... + if (fMergeContext.fDestinationEntryIndex > 0) { + fDestination.CommitCluster(fMergeContext.fDestinationEntryIndex); + } + } +}; +} // namespace ROOT::Experimental::Internal + +ROOT::RResult RNTupleMergeStrategySlow::MergeSource(RPageSource &source, RDescriptorsComparison &descCmp) +{ + const auto &dstDesc = fDestination.GetDescriptor(); + + // Make sure that types of common fields match + for (const auto &[srcField, dstField] : descCmp.fCommonFields) { + const auto srcType = fSrcDescriptor->GetTypeNameForComparison(*srcField); + const auto dstType = dstDesc.GetTypeNameForComparison(*dstField); + if (srcType != dstType) { + return R__FAIL("Field `" + fSrcDescriptor->GetQualifiedFieldName(srcField->GetId()) + + "` has a type incompatible with a previously-seen field with the same name: (old: " + srcType + + ", new: " + dstType + ")"); + } + } + + // Impose the destination model on the source. This will ensure we properly schema-evolve the fields if + // necessary (e.g. if this source has fewer/reordered fields than the destination) + auto srcModel = fDstModel->Clone(); + + // NOTE: this code is almost the same as RNTupleReader::ConnectModel(). + // Perhaps it can be made common? + { + auto &fieldZero = ROOT::Internal::GetFieldZeroOfModel(*srcModel); + ROOT::Internal::SetAllowFieldSubstitutions(fieldZero, true); + // We must not use the descriptor guard to prevent recursive locking in field.ConnectPageSource + ROOT::DescriptorId_t fieldZeroId = fSrcDescriptor->GetFieldZeroId(); + fieldZero.SetOnDiskId(fieldZeroId); + const auto resetFieldIdAndColumnRepresentatives = [](ROOT::RFieldBase &field, + const auto &recursiveCall) -> void { + field.SetOnDiskId(ROOT::kInvalidDescriptorId); + field.SetColumnRepresentatives({}); + for (auto *f : field.GetMutableSubfields()) { + recursiveCall(*f, recursiveCall); + } + }; + // Iterate only over fieldZero's direct subfields; their descendants are recursively handled in + // RFieldBase::ConnectPageSource + for (auto *field : fieldZero.GetMutableSubfields()) { + // We need to reset column reprs otherwise ConnectPageSource fails in case we are changing the column reprs + // due to AutoAdjustColumnRepresentatives. + // We also need to reset on disk ids otherwise fields get matched with the wrong ones using the writer's ids + // (which differ from the reader's) + resetFieldIdAndColumnRepresentatives(*field, resetFieldIdAndColumnRepresentatives); + + if (field->GetOnDiskId() == ROOT::kInvalidDescriptorId) + field->SetOnDiskId(fSrcDescriptor->FindFieldId(field->GetFieldName(), fieldZeroId)); + + // If the OnDiskId is still invalid at this point it means this is an extraDstField + // and we shouldn't/cannot connect it. + if (field->GetOnDiskId() != ROOT::kInvalidDescriptorId) + ROOT::Internal::CallConnectPageSourceOnField(*field, source); + else + ROOT::Internal::CallSetArtificialOnField(*field); + } + ROOT::Internal::SetAllowFieldSubstitutions(fieldZero, false); + } + + auto srcEntry = srcModel->CreateBareEntry(); + + // Create and connect fields to source and sink and bind them together. + // This causes a read on the src entry to populate the same memory used by the dst entry to write. + for (const auto &[srcFieldDesc, _dstFieldDesc] : descCmp.fCommonFields) { + srcEntry->BindValue(srcFieldDesc->GetFieldName(), fDstEntry->GetPtr(srcFieldDesc->GetFieldName())); + } + // reset the extraDstfields, to make sure any subfield that's missing from the current source + // is default-initialized (rather than keeping its previous value). + for (const auto *field : descCmp.fExtraDstFields) { + fDstEntry->EmplaceNewValue(field->GetFieldName()); + } + + // Do the actual data reading/writing + for (auto idx = 0u; idx < fSrcDescriptor->GetNEntries(); ++idx) { + srcEntry->Read(idx); + fMergeContext.fSourceEntryIndex = idx; + while (true) { + auto res = fActions->OnFill(*fDstEntry, fMergeContext); + if (res == ENTupleMergeActionResult::kDrop) + break; + fMergeContext.fDestinationEntryIndex += 1; + fDstEntry->Append(); + if (res == ENTupleMergeActionResult::kEmit) + break; + } + } + + if (fMergeContext.fDestinationEntryIndex > 0) { + for (auto &field : ROOT::Internal::GetFieldZeroOfModel(*fDstModel)) { + ROOT::Internal::CallFlushColumnsOnField(field); + } + } + + return ROOT::RResult::Success(); +} + +// ========================================================================================= +// +// RNTupleMerger +// +// ========================================================================================= RNTupleMerger::RNTupleMerger(std::unique_ptr destination, std::unique_ptr model) : fDestination(std::move(destination)), fModel(std::move(model)) @@ -1534,6 +1737,9 @@ ROOT::RResult RNTupleMerger::Merge(std::span sources, const assert(fDestination); + // TEMP + constexpr bool doSlowMerge = true; + // Set compression settings if unset and verify it's compatible with the sink { const auto dstCompSettings = fDestination->GetWriteOptions().GetCompression(); @@ -1554,6 +1760,10 @@ ROOT::RResult RNTupleMerger::Merge(std::span sources, const "merging) can only be done by providing a valid ROOT::RNTupleModel when constructing the RNTupleMerger."); } + if (doSlowMerge && fModel) { + return R__FAIL("incremental slow merging is currently unsupported."); + } + // NOTE: don't wrap this in a do {} while (0)! It uses continue! #define SKIP_OR_ABORT(errMsg) \ if (mergeOpts.fErrBehavior == ENTupleMergeErrBehavior::kSkip) { \ @@ -1564,7 +1774,9 @@ ROOT::RResult RNTupleMerger::Merge(std::span sources, const } auto strat = - std::unique_ptr(new RNTupleMergeStrategyDefault(fModel, *fDestination, mergeOpts)); + doSlowMerge + ? std::unique_ptr(new RNTupleMergeStrategySlow(fModel, *fDestination, mergeOpts)) + : std::unique_ptr(new RNTupleMergeStrategyDefault(fModel, *fDestination, mergeOpts)); // Merge main loop for (RPageSource *source : sources) { From 03519ffca1f0cf043e65f56306355cbf53b11205 Mon Sep 17 00:00:00 2001 From: silverweed Date: Mon, 5 Oct 2026 10:03:22 +0200 Subject: [PATCH 4/5] [ntuple] Properly handle projected fields in slow merging --- tree/ntuple/src/RNTupleMerger.cxx | 69 +++++++++++++++++++++---------- 1 file changed, 48 insertions(+), 21 deletions(-) diff --git a/tree/ntuple/src/RNTupleMerger.cxx b/tree/ntuple/src/RNTupleMerger.cxx index c145cf08c4e8d..58a01c0da438c 100644 --- a/tree/ntuple/src/RNTupleMerger.cxx +++ b/tree/ntuple/src/RNTupleMerger.cxx @@ -550,6 +550,31 @@ FillCommonAndExtraFields(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTuple return res; } +static void VerifyProjectedFieldsCompatibility(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTupleDescriptor &src, + const ROOT::RFieldDescriptor &dstField, + const ROOT::RFieldDescriptor &srcField, std::vector &errors) +{ + bool projCompatible = srcField.IsProjectedField() == dstField.IsProjectedField(); + if (!projCompatible) { + std::stringstream ss; + ss << "Field `" << dstField.GetFieldName() + << "` is incompatible with previously-seen field with that name because the " + << (srcField.IsProjectedField() ? "new" : "old") << " one is projected and the other isn't"; + errors.push_back(ss.str()); + } else if (srcField.IsProjectedField()) { + // if both fields are projected, verify that they point to the same real field + const auto srcName = src.GetQualifiedFieldName(srcField.GetProjectionSourceId()); + const auto dstName = dst.GetQualifiedFieldName(dstField.GetProjectionSourceId()); + if (srcName != dstName) { + std::stringstream ss; + ss << "Field `" << dstField.GetFieldName() + << "` is projected to a different field than a previously-seen field with the same name (old: " << dstName + << ", new: " << srcName << ")"; + errors.push_back(ss.str()); + } + } +} + /// Compares the top level fields of `dst` and `src` and determines whether they can be merged or not. /// In addition, returns the differences between `dst` and `src`'s structures static ROOT::RResult @@ -574,24 +599,7 @@ CompareDescriptorStructure(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTup const auto &fieldName = field.fSrc->GetFieldName(); // Require that fields are both projected or both not projected - bool projCompatible = field.fSrc->IsProjectedField() == field.fDst->IsProjectedField(); - if (!projCompatible) { - std::stringstream ss; - ss << "Field `" << fieldName << "` is incompatible with previously-seen field with that name because the " - << (field.fSrc->IsProjectedField() ? "new" : "old") << " one is projected and the other isn't"; - errors.push_back(ss.str()); - } else if (field.fSrc->IsProjectedField()) { - // if both fields are projected, verify that they point to the same real field - const auto srcName = src.GetQualifiedFieldName(field.fSrc->GetProjectionSourceId()); - const auto dstName = dst.GetQualifiedFieldName(field.fDst->GetProjectionSourceId()); - if (srcName != dstName) { - std::stringstream ss; - ss << "Field `" << fieldName - << "` is projected to a different field than a previously-seen field with the same name (old: " - << dstName << ", new: " << srcName << ")"; - errors.push_back(ss.str()); - } - } + VerifyProjectedFieldsCompatibility(dst, src, *field.fDst, *field.fSrc, errors); // Require that fields types match // TODO(gparolini): allow non-identical but compatible types @@ -1576,6 +1584,7 @@ class RNTupleMergeStrategySlow final : public RNTupleMergeStrategy { if (!fDstModel) { ROOT::RNTupleDescriptor::RCreateModelOptions cmOpts; cmOpts.SetCreateBare(true); + cmOpts.SetReconstructProjections(true); fDstModel = source.GetSharedDescriptorGuard()->CreateModel(cmOpts); fDestination.Init(*fDstModel); fDstEntry = fDstModel->CreateEntry(); @@ -1584,7 +1593,22 @@ class RNTupleMergeStrategySlow final : public RNTupleMergeStrategy { ROOT::RResult CompareSrcAndDstDescriptors() final { - return FillCommonAndExtraFields(fDstDescriptor, *fSrcDescriptor); + auto res = FillCommonAndExtraFields(fDstDescriptor, *fSrcDescriptor); + + // Verify that projected fields are consistent + std::vector errors; + for (const auto &[srcF, dstF] : res.fCommonFields) { + VerifyProjectedFieldsCompatibility(fDstDescriptor, *fSrcDescriptor, *dstF, *srcF, errors); + } + if (!errors.empty()) { + std::string errMsg = "Schemas of merged RNTuples is incompatible:"; + for (const auto &err : errors) { + errMsg += std::string("\n * ") + err; + } + return R__FAIL(errMsg); + } + + return res; } [[nodiscard]] ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) final @@ -1682,12 +1706,15 @@ ROOT::RResult RNTupleMergeStrategySlow::MergeSource(RPageSource &source, R // Create and connect fields to source and sink and bind them together. // This causes a read on the src entry to populate the same memory used by the dst entry to write. for (const auto &[srcFieldDesc, _dstFieldDesc] : descCmp.fCommonFields) { - srcEntry->BindValue(srcFieldDesc->GetFieldName(), fDstEntry->GetPtr(srcFieldDesc->GetFieldName())); + if (!srcFieldDesc->IsProjectedField()) + srcEntry->BindValue(srcFieldDesc->GetFieldName(), fDstEntry->GetPtr(srcFieldDesc->GetFieldName())); } // reset the extraDstfields, to make sure any subfield that's missing from the current source // is default-initialized (rather than keeping its previous value). for (const auto *field : descCmp.fExtraDstFields) { - fDstEntry->EmplaceNewValue(field->GetFieldName()); + if (!field->IsProjectedField()) { + fDstEntry->EmplaceNewValue(field->GetFieldName()); + } } // Do the actual data reading/writing From 8cfc0e73e7837dd164815ef39214990cc889312d Mon Sep 17 00:00:00 2001 From: silverweed Date: Fri, 23 Oct 2026 10:34:59 +0200 Subject: [PATCH 5/5] [ntuple] Change CompareDescriptors to discriminate slow/default mergeability --- tree/ntuple/src/RNTupleMerger.cxx | 70 ++++++++++++++----------------- 1 file changed, 31 insertions(+), 39 deletions(-) diff --git a/tree/ntuple/src/RNTupleMerger.cxx b/tree/ntuple/src/RNTupleMerger.cxx index 58a01c0da438c..11df28b565ed7 100644 --- a/tree/ntuple/src/RNTupleMerger.cxx +++ b/tree/ntuple/src/RNTupleMerger.cxx @@ -414,6 +414,10 @@ struct RDescriptorsComparison { std::vector fExtraDstFields; std::vector fExtraSrcFields; std::vector fCommonFields; + + // Empty if and only if the source can be fast merged. + std::string fFastMergeErrors; + // For each field that has more than 1 column representation in the output model, // maps the column representatives of the source field with those of the destination. // The key is the destination field. @@ -587,10 +591,13 @@ CompareDescriptorStructure(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTup // 4. dst and src have fields that differ (compatible or incompatible) RDescriptorsComparison res = FillCommonAndExtraFields(dst, src); + std::vector slowMergeErrors; std::vector errors; // Check compatibility of common fields auto fieldsToCheck = res.fCommonFields; + const auto nTopLevelFields = fieldsToCheck.size(); + // NOTE: using index-based for loop because the collection may get extended by the iteration for (std::size_t fieldIdx = 0; fieldIdx < fieldsToCheck.size(); ++fieldIdx) { const auto &field = fieldsToCheck[fieldIdx]; @@ -599,7 +606,8 @@ CompareDescriptorStructure(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTup const auto &fieldName = field.fSrc->GetFieldName(); // Require that fields are both projected or both not projected - VerifyProjectedFieldsCompatibility(dst, src, *field.fDst, *field.fSrc, errors); + auto &errs = (fieldIdx < nTopLevelFields) ? slowMergeErrors : errors; + VerifyProjectedFieldsCompatibility(dst, src, *field.fDst, *field.fSrc, errs); // Require that fields types match // TODO(gparolini): allow non-identical but compatible types @@ -680,14 +688,22 @@ CompareDescriptorStructure(const ROOT::RNTupleDescriptor &dst, const ROOT::RNTup } std::string errMsg; - for (const auto &err : errors) - errMsg += std::string("\n * ") + err; + if (!slowMergeErrors.empty()) { + // If there are slow merge errors, the source cannot be merged at all. + for (const auto &err : slowMergeErrors) + errMsg += std::string("\n * ") + err; - if (!errMsg.empty()) errMsg = errMsg.substr(1); // strip initial newline - - if (errMsg.length()) return R__FAIL(errMsg); + } + + // Source can only be slow merged. + if (!errors.empty()) { + for (const auto &err : errors) + errMsg += std::string("\n * ") + err; + errMsg = errMsg.substr(1); // strip initial newline + res.fFastMergeErrors = errMsg; + } return ROOT::RResult(res); } @@ -967,10 +983,6 @@ class RNTupleMergeStrategy { // Invoked as the first thing when we start processing a new source virtual void InitSource(RPageSource &source) = 0; - // Invoked before merging starts, right after initting the source. Determines whether the current source's schema is - // compatible with the destination's. - [[nodiscard]] virtual ROOT::RResult CompareSrcAndDstDescriptors() = 0; - // Invoked when the destination model must be late model extended (only invoked in case of union merging) [[nodiscard]] virtual ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) = 0; @@ -1046,11 +1058,6 @@ class RNTupleMergeStrategyDefault final : public RNTupleMergeStrategy { } } - ROOT::RResult CompareSrcAndDstDescriptors() final - { - return CompareDescriptorStructure(fDstDescriptor, *fSrcDescriptor); - } - [[nodiscard]] ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) final { // late model extension for all fExtraSrcFields in Union mode @@ -1591,26 +1598,6 @@ class RNTupleMergeStrategySlow final : public RNTupleMergeStrategy { } } - ROOT::RResult CompareSrcAndDstDescriptors() final - { - auto res = FillCommonAndExtraFields(fDstDescriptor, *fSrcDescriptor); - - // Verify that projected fields are consistent - std::vector errors; - for (const auto &[srcF, dstF] : res.fCommonFields) { - VerifyProjectedFieldsCompatibility(fDstDescriptor, *fSrcDescriptor, *dstF, *srcF, errors); - } - if (!errors.empty()) { - std::string errMsg = "Schemas of merged RNTuples is incompatible:"; - for (const auto &err : errors) { - errMsg += std::string("\n * ") + err; - } - return R__FAIL(errMsg); - } - - return res; - } - [[nodiscard]] ROOT::RResult ExtendDestinationModel(RDescriptorsComparison &descCmp) final { // late model extension for all fExtraSrcFields in Union mode @@ -1764,8 +1751,8 @@ ROOT::RResult RNTupleMerger::Merge(std::span sources, const assert(fDestination); - // TEMP - constexpr bool doSlowMerge = true; + // TODO(gparolini): this is here to allow testing of the slow merging path, it will disappear in the future. + constexpr bool doSlowMerge = false; // Set compression settings if unset and verify it's compatible with the sink { @@ -1808,7 +1795,6 @@ ROOT::RResult RNTupleMerger::Merge(std::span sources, const // Merge main loop for (RPageSource *source : sources) { strat->InitSource(*source); - { auto srcDescriptor = source->GetSharedDescriptorGuard(); if (srcDescriptor->GetVersion() > ROOT::RNTuple::GetCurrentVersion()) { @@ -1828,12 +1814,18 @@ ROOT::RResult RNTupleMerger::Merge(std::span sources, const fDestination->UpdateExtraTypeInfo(extraTypeInfoDesc); } - auto descCmpRes = strat->CompareSrcAndDstDescriptors(); + auto descCmpRes = + CompareDescriptorStructure(fDestination->GetDescriptor(), source->GetSharedDescriptorGuard().GetConstRef()); if (!descCmpRes) { SKIP_OR_ABORT(std::string("Source RNTuple has an incompatible schema with the destination:\n") + descCmpRes.GetError()->GetReport()) } auto descCmp = descCmpRes.Unwrap(); + // TODO: switch to slow merge if we cannot "fast" merge + if (!doSlowMerge && !descCmp.fFastMergeErrors.empty()) { + SKIP_OR_ABORT(std::string("Source RNTuple has an incompatible schema with the destination:\n") + + descCmp.fFastMergeErrors) + } // If the current source is missing some fields and we're not in Union mode, error // (if we are in Union mode, MergeSourceClusters will fill the missing fields with default values).