YT-22068: TError-ize statistics merge

- Return `TError` on `Merge`, `MergeWithOverride`, `AppendStatistics`, `AppendTaggedSummary`

- Do not reset “good” statistics if merge fails,  add a test to verify this behavior
commit_hash:2c3f30c33b82defe4f59192724487fca29b63383

4 files changed, 69 insertions, 33 deletions

diff --git a/yt/yt/core/misc/statistics-inl.h b/yt/yt/core/misc/statistics-inl.h
index d90add25e8..cf71e18086 100644
--- a/yt/yt/core/misc/statistics-inl.h
+++ b/yt/yt/core/misc/statistics-inl.h
@@ -66,14 +66,14 @@ std::pair<EStatisticPathConflictType, typename TSummaryMap::iterator> IsCompatib
 
 //! Tries to emplace statistic into TSummaryMap, and checks if it is valid and compatible.
 template <typename TSummaryMap, typename... Ts>
-std::pair<typename TSummaryMap::iterator, bool> CheckedEmplaceStatistic(
+TErrorOr<std::pair<typename TSummaryMap::iterator, bool>> CheckedEmplaceStatistic(
     TSummaryMap& existingStatistics,
     const NStatisticPath::TStatisticPath& path,
     Ts&&... args)
 {
     auto [conflictType, hintIt] = IsCompatibleStatistic(existingStatistics, path);
     if (conflictType == EStatisticPathConflictType::Exists) {
-        return {hintIt, false};
+        return std::pair{hintIt, false};
     }
     if (conflictType != EStatisticPathConflictType::None) {
         auto prefixPath = hintIt->first;
@@ -83,21 +83,28 @@ std::pair<typename TSummaryMap::iterator, bool> CheckedEmplaceStatistic(
             std::swap(prefixPath, conflictPath);
         }
 
-        THROW_ERROR_EXCEPTION("Statistic path cannot be a prefix of another statistic path")
+        return TError("Statistic path cannot be a prefix of another statistic path")
             << TErrorAttribute("prefix_path", prefixPath)
             << TErrorAttribute("contained_in_path", conflictPath);
     }
     auto emplacedIt = existingStatistics.emplace_hint(hintIt, path, std::forward<Ts>(args)...);
-    return {emplacedIt, true};
+    return std::pair{emplacedIt, true};
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
 template <class TTags>
-void TTaggedStatistics<TTags>::AppendStatistics(const TStatistics& statistics, TTags tags)
+TError TTaggedStatistics<TTags>::AppendStatistics(const TStatistics& statistics, TTags tags)
 {
+    TError error;
     for (const auto& [path, summary] : statistics.Data()) {
-        auto [emplacedIterator, _] = CheckedEmplaceStatistic(Data_, path, TTaggedSummaries{});
+        auto emplaceResult = CheckedEmplaceStatistic(Data_, path, TTaggedSummaries{});
+        if (!emplaceResult.IsOK()) {
+            error = emplaceResult;
+            continue;
+        }
+
+        auto [emplacedIterator, _] = emplaceResult.Value();
         auto& pathSummaries = emplacedIterator->second;
         auto it = pathSummaries.find(tags);
         if (it == pathSummaries.end()) {
@@ -106,14 +113,20 @@ void TTaggedStatistics<TTags>::AppendStatistics(const TStatistics& statistics, T
             it->second.Merge(summary);
         }
     }
+    return error;
 }
 
 template <class TTags>
-void TTaggedStatistics<TTags>::AppendTaggedSummary(const NStatisticPath::TStatisticPath& path, const TTaggedStatistics<TTags>::TTaggedSummaries& taggedSummaries)
+TError TTaggedStatistics<TTags>::AppendTaggedSummary(const NStatisticPath::TStatisticPath& path, const TTaggedStatistics<TTags>::TTaggedSummaries& taggedSummaries)
 {
-    auto [taggedSummariesIt, emplaceHappened] = CheckedEmplaceStatistic(Data_, path, taggedSummaries);
+    auto emplaceResult = CheckedEmplaceStatistic(Data_, path, taggedSummaries);
+    if (!emplaceResult.IsOK()) {
+        return TError(emplaceResult);
+    }
+
+    auto [taggedSummariesIt, emplaceHappened] = emplaceResult.Value();
     if (emplaceHappened) {
-        return;
+        return {};
     }
 
     auto& currentTaggedSummaries = taggedSummariesIt->second;
@@ -124,6 +137,7 @@ void TTaggedStatistics<TTags>::AppendTaggedSummary(const NStatisticPath::TStatis
             summaryIt->second.Merge(summary);
         }
     }
+    return {};
 }
 
 template <class TTags>
diff --git a/yt/yt/core/misc/statistics.cpp b/yt/yt/core/misc/statistics.cpp
index 0181964243..d853716cfd 100644
--- a/yt/yt/core/misc/statistics.cpp
+++ b/yt/yt/core/misc/statistics.cpp
@@ -93,15 +93,21 @@ bool TSummary::operator ==(const TSummary& other) const
 
 ////////////////////////////////////////////////////////////////////////////////
 
-TSummary& TStatistics::GetSummary(const TStatisticPath& path)
+TErrorOr<TSummary*> TStatistics::GetSummary(const TStatisticPath& path)
 {
-    auto [it, _] = CheckedEmplaceStatistic(Data_, path, TSummary());
-    return it->second;
+    auto emplaceResult = CheckedEmplaceStatistic(Data_, path, TSummary());
+    if (!emplaceResult.IsOK()) {
+        return TError(emplaceResult);
+    }
+    auto [it, _] = emplaceResult.Value();
+    return &it->second;
 }
 
 void TStatistics::AddSample(const TStatisticPath& path, i64 sample)
 {
-    GetSummary(path).AddSample(sample);
+    auto summary = GetSummary(path);
+    THROW_ERROR_EXCEPTION_IF(!summary.IsOK(), summary);
+    summary.Value()->AddSample(sample);
 }
 
 void TStatistics::AddSample(const TStatisticPath& path, const INodePtr& sample)
@@ -113,9 +119,10 @@ void TStatistics::AddSample(const TStatisticPath& path, const INodePtr& sample)
 
 void TStatistics::ReplacePathWithSample(const TStatisticPath& path, const i64 sample)
 {
-    auto& summary = GetSummary(path);
-    summary.Reset();
-    summary.AddSample(sample);
+    auto summary = GetSummary(path);
+    THROW_ERROR_EXCEPTION_IF(!summary.IsOK(), summary);
+    summary.Value()->Reset();
+    summary.Value()->AddSample(sample);
 }
 
 void TStatistics::ReplacePathWithSample(const TStatisticPath& path, const INodePtr& sample)
@@ -151,17 +158,32 @@ void TStatistics::ProcessNodeWithCallback(const TStatisticPath& path, const NYTr
     }
 }
 
-void TStatistics::Merge(const TStatistics& statistics)
+TError TStatistics::Merge(const TStatistics& statistics)
 {
+    TError error;
     for (const auto& [path, summary] : statistics.Data()) {
-        GetSummary(path).Merge(summary);
+        auto currentSummary = GetSummary(path);
+        if (currentSummary.IsOK()) {
+            currentSummary.Value()->Merge(summary);
+        } else {
+            error = currentSummary;
+        }
     }
+    return error;
 }
 
-void TStatistics::MergeWithOverride(const TStatistics& statistics)
+TError TStatistics::MergeWithOverride(const TStatistics& statistics)
 {
-    const auto& otherData = statistics.Data();
-    Data_.insert(otherData.begin(), otherData.end());
+    TError error;
+    for (const auto& [path, summary] : statistics.Data()) {
+        auto currentSummary = GetSummary(path);
+        if (currentSummary.IsOK()) {
+            *currentSummary.Value() = summary;
+        } else {
+            error = currentSummary;
+        }
+    }
+    return error;
 }
 
 // TODO(pavook) quite funky implementation details. Should we move this into statistic_path.h?
diff --git a/yt/yt/core/misc/statistics.h b/yt/yt/core/misc/statistics.h
index 606992cfc4..776c420a9c 100644
--- a/yt/yt/core/misc/statistics.h
+++ b/yt/yt/core/misc/statistics.h
@@ -73,9 +73,9 @@ public:
     void ReplacePathWithSample(const NStatisticPath::TStatisticPath& path, const T& sample);
 
     //! Merge statistics by merging summaries for each common statistics path.
-    void Merge(const TStatistics& statistics);
+    TError Merge(const TStatistics& statistics);
     //! Merge statistics by taking summary from #statistics for each common statistics path.
-    void MergeWithOverride(const TStatistics& statistics);
+    TError MergeWithOverride(const TStatistics& statistics);
 
     //! Get range of all elements whose path starts with a given strict prefix path (possibly empty).
     /*!
@@ -94,7 +94,7 @@ private:
     template <class TCallback>
     void ProcessNodeWithCallback(const NStatisticPath::TStatisticPath& path, const NYTree::INodePtr& sample, TCallback callback);
 
-    TSummary& GetSummary(const NStatisticPath::TStatisticPath& path);
+    TErrorOr<TSummary*> GetSummary(const NStatisticPath::TStatisticPath& path);
 
     friend class TStatisticsBuildingConsumer;
 };
@@ -144,8 +144,8 @@ public:
     using TTaggedSummaries = THashMap<TTags, TSummary>;
     using TSummaryMap = std::map<NStatisticPath::TStatisticPath, TTaggedSummaries>;
 
-    void AppendStatistics(const TStatistics& statistics, TTags tags);
-    void AppendTaggedSummary(const NStatisticPath::TStatisticPath& path, const TTaggedSummaries& taggedSummaries);
+    TError AppendStatistics(const TStatistics& statistics, TTags tags);
+    TError AppendTaggedSummary(const NStatisticPath::TStatisticPath& path, const TTaggedSummaries& taggedSummaries);
 
     const TTaggedSummaries* FindTaggedSummaries(const NStatisticPath::TStatisticPath& path) const;
     const TSummaryMap& GetData() const;
diff --git a/yt/yt/core/misc/unittests/statistics_ut.cpp b/yt/yt/core/misc/unittests/statistics_ut.cpp
index 12955b2017..7f4a138c4d 100644
--- a/yt/yt/core/misc/unittests/statistics_ut.cpp
+++ b/yt/yt/core/misc/unittests/statistics_ut.cpp
@@ -90,14 +90,14 @@ TEST(TStatistics, AddSample)
 
     statistics.Merge(CreateStatistics({
         {"key"_L / "subkey"_L / "x"_L, 5},
-        {"key"_L / "subkey"_L / "z"_L, 9}}));
+        {"key"_L / "subkey"_L / "z"_L, 9}})).ThrowOnError();
 
     EXPECT_EQ(10, GetNumericValue(statistics, "key"_L / "subkey"_L / "x"_L));
     EXPECT_EQ(7, GetNumericValue(statistics, "key"_L / "subkey"_L / "y"_L));
     EXPECT_EQ(9, GetNumericValue(statistics, "key"_L / "subkey"_L / "z"_L));
 
     EXPECT_THROW(
-        statistics.Merge(CreateStatistics({{"key"_L, 5}})),
+        statistics.Merge(CreateStatistics({{"key"_L, 5}})).ThrowOnError(),
         std::exception);
 
     statistics.AddSample("key"_L / "subkey"_L / "x"_L, 10);
@@ -231,20 +231,20 @@ TEST(TTaggedStatistics, AppendStatistics)
         statistics.AddSample("abc"_L / "def"_L, 1);
         statistics.AddSample("abc"_L / "defg"_L, 2);
         statistics.AddSample("xyz"_L, 3);
-        taggedStatistics.AppendStatistics(statistics, 1);
+        taggedStatistics.AppendStatistics(statistics, 1).ThrowOnError();
     }
 
     {
         TStatistics statistics;
         statistics.AddSample("abc"_L / "def"_L, 1);
         statistics.AddSample("ijk"_L, 2);
-        taggedStatistics.AppendStatistics(statistics, 2);
+        taggedStatistics.AppendStatistics(statistics, 2).ThrowOnError();
     }
 
     {
         TStatistics statistics;
         statistics.AddSample("abc"_L / "def"_L, 2);
-        taggedStatistics.AppendStatistics(statistics, 1);
+        taggedStatistics.AppendStatistics(statistics, 1).ThrowOnError();
     }
 
     {
@@ -269,13 +269,13 @@ TEST(TTaggedStatistics, AppendStatistics)
     {
         TStatistics statistics;
         statistics.AddSample("xyz"_L / "suffix"_L, 1);
-        EXPECT_THROW(taggedStatistics.AppendStatistics(statistics, 3), std::exception);
+        EXPECT_THROW(taggedStatistics.AppendStatistics(statistics, 3).ThrowOnError(), std::exception);
     }
 
     {
         TStatistics statistics;
         statistics.AddSample("abc"_L, 1); // prefix
-        EXPECT_THROW(taggedStatistics.AppendStatistics(statistics, 3), std::exception);
+        EXPECT_THROW(taggedStatistics.AppendStatistics(statistics, 3).ThrowOnError(), std::exception);
     }
 }