YQL-17250: Add operation information into hybrid statistics (#597)

mxkovalev · web-flow · commit 3facfc056204 · 2023-12-21T15:13:28.000+03:00
* YQL-17250: Add operation information into hybrid statistics

* Fix review comments
diff --git a/ydb/library/yql/providers/common/provider/yql_provider.cpp b/ydb/library/yql/providers/common/provider/yql_provider.cpp
@@ -1264,14 +1264,16 @@ void WriteStatistics(NYson::TYsonWriter& writer, const TOperationStatistics& sta
     writer.OnEndMap();
 }
 
-void WriteStatistics(NYson::TYsonWriter& writer, bool totalOnly, const THashMap<ui32, TOperationStatistics>& statistics) {
+void WriteStatistics(NYson::TYsonWriter& writer, bool totalOnly, const THashMap<ui32, TOperationStatistics>& statistics, bool addExternalMap) {
     if (statistics.empty()) {
         return;
     }
 
     THashMap<TString, std::tuple<i64, i64, i64, TMaybe<i64>>> total; // sum, count, max, min
 
-    writer.OnBeginMap();
+    if (addExternalMap) {
+        writer.OnBeginMap();
+    }
 
     for (const auto& opStatistics : statistics) {
         for (auto& el : opStatistics.second.Entries) {
@@ -1331,8 +1333,9 @@ void WriteStatistics(NYson::TYsonWriter& writer, bool totalOnly, const THashMap<
         writer.OnEndMap();
     }
     writer.OnEndMap(); // total
-
-    writer.OnEndMap();
+    if (addExternalMap) {
+        writer.OnEndMap();
+    }
 }
 
 bool ValidateCompressionForInput(std::string_view format, std::string_view compression, TExprContext& ctx) {
diff --git a/ydb/library/yql/providers/common/provider/yql_provider.h b/ydb/library/yql/providers/common/provider/yql_provider.h
@@ -179,7 +179,7 @@ void WriteStreams(NYson::TYsonWriter& writer, TStringBuf name, const NNodes::TCo
 
 double GetDataReplicationFactor(const TExprNode& lambda, TExprContext& ctx);
 
-void WriteStatistics(NYson::TYsonWriter& writer, bool totalOnly, const THashMap<ui32, TOperationStatistics>& statistics);
+void WriteStatistics(NYson::TYsonWriter& writer, bool totalOnly, const THashMap<ui32, TOperationStatistics>& statistics, bool addExternalMap = true);
 void WriteStatistics(NYson::TYsonWriter& writer, const TOperationStatistics& statistics);
 
 bool ValidateCompressionForInput(std::string_view format, std::string_view compression, TExprContext& ctx);
diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_datasink.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_datasink.cpp
@@ -160,7 +160,16 @@ class TYtDataSink : public TDataProviderBase {
             return false;
         }
 
-        NCommon::WriteStatistics(writer, totalOnly, State_->Statistics);
+        writer.OnBeginMap();
+            NCommon::WriteStatistics(writer, totalOnly, State_->Statistics, false);
+            writer.OnKeyedItem("Hybrid");
+            writer.OnBeginMap();
+                for (const auto& [opName, stats] : State_->HybridStatistics) {
+                    writer.OnKeyedItem(opName);
+                    NCommon::WriteStatistics(writer, totalOnly, {{0, stats}});
+                }
+            writer.OnEndMap();
+        writer.OnEndMap();
 
         return true;
     }
diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_datasink_exec.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_datasink_exec.cpp
@@ -57,21 +57,23 @@ bool NeedFallback(const TIssues& issues) {
 
 TIssue WrapIssuesOnHybridFallback(TPosition pos, const TIssues& issues) {
     TIssue result(pos, "Hybrid execution fallback on YT");
-    result.SetCode(TIssuesIds::DQ_GATEWAY_NEED_FALLBACK_ERROR, TSeverityIds::S_WARNING);
+    result.SetCode(TIssuesIds::DQ_GATEWAY_NEED_FALLBACK_ERROR, TSeverityIds::S_INFO);
 
-    const std::function<void(TIssue& issue)> toWarning = [&](TIssue& issue) {
-        if (issue.Severity == TSeverityIds::S_ERROR || issue.Severity == TSeverityIds::S_FATAL) {
-            issue.Severity = TSeverityIds::S_WARNING;
+    const std::function<void(TIssue& issue)> toInfo = [&](TIssue& issue) {
+        if (issue.Severity == TSeverityIds::S_ERROR
+            || issue.Severity == TSeverityIds::S_FATAL
+            || issue.Severity == TSeverityIds::S_WARNING) {
+            issue.Severity = TSeverityIds::S_INFO;
         }
         for (const auto& subissue : issue.GetSubIssues()) {
-            toWarning(*subissue);
+            toInfo(*subissue);
         }
     };
 
     for (const auto& issue : issues) {
-        TIssuePtr warning(new TIssue(issue));
-        toWarning(*warning);
-        result.AddSubIssue(std::move(warning));
+        TIssuePtr info(new TIssue(issue));
+        toInfo(*info);
+        result.AddSubIssue(std::move(info));
     }
 
     return result;
@@ -125,6 +127,11 @@ class TYtDataSinkExecTransformer : public TExecTransformerBase {
     static TExprNode::TPtr FinalizeOutputOp(const TYtState::TPtr& state, const TString& operationHash,
         const IYtGateway::TRunResult& res, const TExprNode::TPtr& input, TExprNode::TPtr& output, TExprContext& ctx, bool markFinished)
     {
+        if (markFinished && !TYtDqProcessWrite::Match(input.Get())) {
+            with_lock(state->StatisticsMutex) {
+                state->HybridStatistics[input->Content()].Entries.emplace_back(TString{"YtExecution"}, 0, 0, 0, 0, 1);
+            }
+        }
         auto outSection = TYtOutputOpBase(input).Output();
         YQL_ENSURE(outSection.Size() == res.OutTableStats.size(), "Invalid output table count in IYtGateway::TRunResult");
         TExprNode::TListType newOutTables;
@@ -286,14 +293,21 @@ class TYtDataSinkExecTransformer : public TExecTransformerBase {
                 State_->Statistics[Max<ui32>()].Entries.emplace_back(TString{name}, 0, 0, 0, 0, 1);
             }
         };
+        auto hybridStatWriter = [this](TStringBuf statName, TStringBuf opName) {
+            with_lock(State_->StatisticsMutex) {
+                State_->HybridStatistics[opName].Entries.emplace_back(TString{statName}, 0, 0, 0, 0, 1);
+            }
+        };
 
         switch (input->Head().GetState()) {
             case TExprNode::EState::ExecutionComplete:
                 statWriter("HybridExecution");
+                hybridStatWriter("Execution", input->TailPtr()->Content());
                 output = input->HeadPtr();
                 break;
             case TExprNode::EState::Error: {
                 statWriter("HybridFallback");
+                hybridStatWriter("Fallback", input->TailPtr()->Content());
                 if (State_->Configuration->HybridDqExecutionFallback.Get().GetOrElse(true)) {
                     output = input->TailPtr();
                 } else {
diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_dq_hybrid.cpp b/ydb/library/yql/providers/yt/provider/yql_yt_dq_hybrid.cpp
@@ -275,10 +275,12 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
                 if (const auto stat = CanReadHybrid(sort.Input().Item(0))) {
                     if (stat->front() <= sizeLimit && stat->back() <= chunksLimit) {
                         YQL_CLOG(INFO, ProviderYt) << "Sort on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";
-                        PushStat("Hybrid_Sort_try");
+                        PushStat("HybridTry");
+                        PushHybridStat("Try", node.Raw()->Content());
                         return MakeYtSortByDq(sort, ctx);
                     }
-                    PushStat("Hybrid_Sort_over_limits");
+                    PushStat("HybridSkipOverLimits");
+                    PushHybridStat("SkipOverLimits", node.Raw()->Content());
                 }
             }
         }
@@ -294,10 +296,12 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
                 if (const auto stat = CanReadHybrid(merge.Input().Item(0))) {
                     if (stat->front() <= sizeLimit && stat->back() <= chunksLimit) {
                         YQL_CLOG(INFO, ProviderYt) << "Merge on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";
-                        PushStat("Hybrid_Merge_try");
+                        PushStat("HybridTry");
+                        PushHybridStat("Try", node.Raw()->Content());
                         return MakeYtSortByDq(merge, ctx);
                     }
-                    PushStat("Hybrid_Merge_over_limits");
+                    PushStat("HybridSkipOverLimits");
+                    PushHybridStat("SkipOverLimits", node.Raw()->Content());
                 }
             }
         }
@@ -347,7 +351,8 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
                 if (stat->front() <= sizeLimit && stat->back() <= chunksLimit) {
                     if (CanExecuteInHybrid(map.Mapper().Ptr(), chunksLimit, sizeLimit)) {
                         YQL_CLOG(INFO, ProviderYt) << "Map on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";
-                        PushStat("Hybrid_Map_try");
+                        PushStat("HybridTry");
+                        PushHybridStat("Try", node.Raw()->Content());
                         TSyncMap syncList;
                         const auto& paths = map.Input().Item(0).Paths();
                         for (auto i = 0U; i < paths.Size(); ++i) {
@@ -430,7 +435,8 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
                             .Done();
                     }
                 }
-                PushStat("Hybrid_Map_over_limits");
+                PushStat("HybridOverLimits");
+                PushHybridStat("SkipOverLimits", node.Raw()->Content());
             }
         }
 
@@ -626,12 +632,14 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
                     if (CanExecuteInHybrid(reduce.Reducer().Ptr(), chunksLimit, sizeLimit)) {
                         if (ETypeAnnotationKind::Struct == GetSeqItemType(*reduce.Reducer().Args().Arg(0).Ref().GetTypeAnn()).GetKind()) {
                             YQL_CLOG(INFO, ProviderYt) << "Reduce on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";
-                            PushStat("Hybrid_Reduce_try");
+                            PushStat("HybridTry");
+                            PushHybridStat("Try", node.Raw()->Content());
                             return MakeYtReduceByDq(reduce, ctx);
                         }
                     }
                 }
-                PushStat("Hybrid_Reduce_over_limits");
+                PushStat("HybridSkipOverLimits");
+                PushHybridStat("SkipOverLimits", node.Raw()->Content());
             }
         }
 
@@ -647,12 +655,14 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
                     if (CanExecuteInHybrid(mapReduce.Reducer().Ptr(), chunksLimit, sizeLimit) && CanExecuteInHybrid(mapReduce.Mapper().Ptr(), chunksLimit, sizeLimit)) {
                         if (ETypeAnnotationKind::Struct == GetSeqItemType(*mapReduce.Reducer().Args().Arg(0).Ref().GetTypeAnn()).GetKind()) {
                             YQL_CLOG(INFO, ProviderYt) << "MapReduce on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";
-                            PushStat("Hybrid_MapReduce_try");
+                            PushHybridStat("Try", node.Raw()->Content());
+                            PushStat("HybridTry");
                             return MakeYtReduceByDq(mapReduce, ctx);
                         }
                     }
                 }
-                PushStat("Hybrid_MapReduce_over_limits");
+                PushHybridStat("SkipOverLimits", node.Raw()->Content());
+                PushStat("HybridOverLimits");
             }
         }
 
@@ -665,6 +675,12 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {
         }
     };
 
+    void PushHybridStat(TStringBuf statName, TStringBuf opName) const {
+        with_lock(State_->StatisticsMutex) {
+            State_->HybridStatistics[opName].Entries.emplace_back(TString{statName}, 0, 0, 0, 0, 1);
+        }
+    };
+
     const TYtState::TPtr State_;
     const THolder<IGraphTransformer> Finalizer_;
 };
diff --git a/ydb/library/yql/providers/yt/provider/yql_yt_provider.h b/ydb/library/yql/providers/yt/provider/yql_yt_provider.h
@@ -96,6 +96,7 @@ struct TYtState : public TThrRefBase {
     THashMap<std::pair<TString, TString>, TString> AnonymousLabels; // cluster + label -> name
     std::unordered_map<ui64, TString> NodeHash; // unique id -> hash
     THashMap<ui32, TOperationStatistics> Statistics; // public id -> stat
+    THashMap<TString, TOperationStatistics> HybridStatistics; // operation name -> stat
     TMutex StatisticsMutex;
     THashSet<std::pair<TString, TString>> Checkpoints; // Set of checkpoint tables
     THolder<IDqIntegration> DqIntegration_;

Original file line number	Diff line number	Diff line change
`@@ -275,10 +275,12 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`275`	`275`	`if (const auto stat = CanReadHybrid(sort.Input().Item(0))) {`
`276`	`276`	`if (stat->front() <= sizeLimit && stat->back() <= chunksLimit) {`
`277`	`277`	`YQL_CLOG(INFO, ProviderYt) << "Sort on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";`
`278`		`- PushStat("Hybrid_Sort_try");`
	`278`	`+ PushStat("HybridTry");`
	`279`	`+ PushHybridStat("Try", node.Raw()->Content());`
`279`	`280`	`return MakeYtSortByDq(sort, ctx);`
`280`	`281`	`}`
`281`		`- PushStat("Hybrid_Sort_over_limits");`
	`282`	`+ PushStat("HybridSkipOverLimits");`
	`283`	`+ PushHybridStat("SkipOverLimits", node.Raw()->Content());`
`282`	`284`	`}`
`283`	`285`	`}`
`284`	`286`	`}`
`@@ -294,10 +296,12 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`294`	`296`	`if (const auto stat = CanReadHybrid(merge.Input().Item(0))) {`
`295`	`297`	`if (stat->front() <= sizeLimit && stat->back() <= chunksLimit) {`
`296`	`298`	`YQL_CLOG(INFO, ProviderYt) << "Merge on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";`
`297`		`- PushStat("Hybrid_Merge_try");`
	`299`	`+ PushStat("HybridTry");`
	`300`	`+ PushHybridStat("Try", node.Raw()->Content());`
`298`	`301`	`return MakeYtSortByDq(merge, ctx);`
`299`	`302`	`}`
`300`		`- PushStat("Hybrid_Merge_over_limits");`
	`303`	`+ PushStat("HybridSkipOverLimits");`
	`304`	`+ PushHybridStat("SkipOverLimits", node.Raw()->Content());`
`301`	`305`	`}`
`302`	`306`	`}`
`303`	`307`	`}`
`@@ -347,7 +351,8 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`347`	`351`	`if (stat->front() <= sizeLimit && stat->back() <= chunksLimit) {`
`348`	`352`	`if (CanExecuteInHybrid(map.Mapper().Ptr(), chunksLimit, sizeLimit)) {`
`349`	`353`	`YQL_CLOG(INFO, ProviderYt) << "Map on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";`
`350`		`- PushStat("Hybrid_Map_try");`
	`354`	`+ PushStat("HybridTry");`
	`355`	`+ PushHybridStat("Try", node.Raw()->Content());`
`351`	`356`	`TSyncMap syncList;`
`352`	`357`	`const auto& paths = map.Input().Item(0).Paths();`
`353`	`358`	`for (auto i = 0U; i < paths.Size(); ++i) {`
`@@ -430,7 +435,8 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`430`	`435`	`.Done();`
`431`	`436`	`}`
`432`	`437`	`}`
`433`		`- PushStat("Hybrid_Map_over_limits");`
	`438`	`+ PushStat("HybridOverLimits");`
	`439`	`+ PushHybridStat("SkipOverLimits", node.Raw()->Content());`
`434`	`440`	`}`
`435`	`441`	`}`
`436`	`442`
`@@ -626,12 +632,14 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`626`	`632`	`if (CanExecuteInHybrid(reduce.Reducer().Ptr(), chunksLimit, sizeLimit)) {`
`627`	`633`	`if (ETypeAnnotationKind::Struct == GetSeqItemType(*reduce.Reducer().Args().Arg(0).Ref().GetTypeAnn()).GetKind()) {`
`628`	`634`	`YQL_CLOG(INFO, ProviderYt) << "Reduce on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";`
`629`		`- PushStat("Hybrid_Reduce_try");`
	`635`	`+ PushStat("HybridTry");`
	`636`	`+ PushHybridStat("Try", node.Raw()->Content());`
`630`	`637`	`return MakeYtReduceByDq(reduce, ctx);`
`631`	`638`	`}`
`632`	`639`	`}`
`633`	`640`	`}`
`634`		`- PushStat("Hybrid_Reduce_over_limits");`
	`641`	`+ PushStat("HybridSkipOverLimits");`
	`642`	`+ PushHybridStat("SkipOverLimits", node.Raw()->Content());`
`635`	`643`	`}`
`636`	`644`	`}`
`637`	`645`
`@@ -647,12 +655,14 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`647`	`655`	`if (CanExecuteInHybrid(mapReduce.Reducer().Ptr(), chunksLimit, sizeLimit) && CanExecuteInHybrid(mapReduce.Mapper().Ptr(), chunksLimit, sizeLimit)) {`
`648`	`656`	`if (ETypeAnnotationKind::Struct == GetSeqItemType(*mapReduce.Reducer().Args().Arg(0).Ref().GetTypeAnn()).GetKind()) {`
`649`	`657`	`YQL_CLOG(INFO, ProviderYt) << "MapReduce on DQ with equivalent input size " << stat->front() << " and " << stat->back() << " chunks.";`
`650`		`- PushStat("Hybrid_MapReduce_try");`
	`658`	`+ PushHybridStat("Try", node.Raw()->Content());`
	`659`	`+ PushStat("HybridTry");`
`651`	`660`	`return MakeYtReduceByDq(mapReduce, ctx);`
`652`	`661`	`}`
`653`	`662`	`}`
`654`	`663`	`}`
`655`		`- PushStat("Hybrid_MapReduce_over_limits");`
	`664`	`+ PushHybridStat("SkipOverLimits", node.Raw()->Content());`
	`665`	`+ PushStat("HybridOverLimits");`
`656`	`666`	`}`
`657`	`667`	`}`
`658`	`668`
`@@ -665,6 +675,12 @@ class TYtDqHybridTransformer : public TOptimizeTransformerBase {`
`665`	`675`	`}`
`666`	`676`	`};`
`667`	`677`
	`678`	`+ void PushHybridStat(TStringBuf statName, TStringBuf opName) const {`
	`679`	`+ with_lock(State_->StatisticsMutex) {`
	`680`	`+ State_->HybridStatistics[opName].Entries.emplace_back(TString{statName}, 0, 0, 0, 0, 1);`
	`681`	`+ }`
	`682`	`+ };`
	`683`	`+`
`668`	`684`	`const TYtState::TPtr State_;`
`669`	`685`	`const THolder<IGraphTransformer> Finalizer_;`
`670`	`686`	`};`