YQ-3738 RD pass UV from filter to read actor (#11940)

Author: GrigoriyPA

ydb/core/fq/libs/row_dispatcher/events/data_plane.h

@@ -8,6 +8,7 @@
 #include <ydb/library/yql/providers/pq/proto/dq_io.pb.h>
 #include <ydb/core/fq/libs/row_dispatcher/events/topic_session_stats.h>
 
+#include <yql/essentials/public/issue/yql_issue.h>
 #include <yql/essentials/public/purecalc/common/fwd.h>
 
 namespace NFq {
@@ -148,9 +149,9 @@ struct TEvRowDispatcher {
     };
 
     struct TEvSessionStatistic : public NActors::TEventLocal<TEvSessionStatistic, EEv::EvSessionStatistic> {
-        TEvSessionStatistic(const TopicSessionStatistic& stat)
+        TEvSessionStatistic(const TTopicSessionStatistic& stat)
         : Stat(stat) {}
-        TopicSessionStatistic Stat;
+        TTopicSessionStatistic Stat;
     };
 
     struct TEvHeartbeat : public NActors::TEventPB<TEvHeartbeat, NFq::NRowDispatcherProto::TEvHeartbeat, EEv::EvHeartbeat> {
@@ -182,16 +183,19 @@ struct TEvRowDispatcher {
     };
 
     struct TEvPurecalcCompileResponse : public NActors::TEventLocal<TEvPurecalcCompileResponse, EEv::EvPurecalcCompileResponse> {
-        explicit TEvPurecalcCompileResponse(const TString& error)
-            : Error(error)
+        TEvPurecalcCompileResponse(NYql::NDqProto::StatusIds::StatusCode status, NYql::TIssues issues)
+            : Status(status)
+            , Issues(std::move(issues))
         {}
 
         explicit TEvPurecalcCompileResponse(IProgramHolder::TPtr programHolder)
             : ProgramHolder(std::move(programHolder))
+            , Status(NYql::NDqProto::StatusIds::SUCCESS)
         {}
 
         IProgramHolder::TPtr ProgramHolder;  // Same holder that passed into TEvPurecalcCompileRequest
-        TString Error;
+        NYql::NDqProto::StatusIds::StatusCode Status;
+        NYql::TIssues Issues;
     };
 };
 

ydb/core/fq/libs/row_dispatcher/events/topic_session_stats.h

@@ -5,7 +5,7 @@
 
 namespace NFq {
 
-struct TopicSessionClientStatistic {
+struct TTopicSessionClientStatistic {
     NActors::TActorId ReadActorId;
     ui32 PartitionId = 0;
     i64 UnreadRows = 0;         // Current value
@@ -15,7 +15,7 @@ struct TopicSessionClientStatistic {
     bool IsWaiting = false;     // Current value
     i64 ReadLagMessages = 0;    // Current value
     ui64 InitialOffset = 0;
-    void Add(const TopicSessionClientStatistic& stat) {
+    void Add(const TTopicSessionClientStatistic& stat) {
         UnreadRows = stat.UnreadRows;
         UnreadBytes = stat.UnreadBytes;
         Offset = stat.Offset;
@@ -29,39 +29,75 @@ struct TopicSessionClientStatistic {
     }
 };
 
-struct TopicSessionCommonStatistic {
+struct TParserStatistic {
+    TDuration ParserLatency;
+
+    void Add(const TParserStatistic& stat) {
+        ParserLatency = stat.ParserLatency != TDuration::Zero() ? stat.ParserLatency : ParserLatency;
+    }
+};
+
+struct TFiltersStatistic {
+    TDuration FilterLatency;
+
+    void Add(const TFiltersStatistic& stat) {
+        FilterLatency = stat.FilterLatency != TDuration::Zero() ? stat.FilterLatency : FilterLatency;
+    }
+};
+
+struct TFormatHandlerStatistic {
+    TDuration ParseAndFilterLatency;
+
+    TParserStatistic ParserStats;
+    TFiltersStatistic FilterStats;
+
+    void Add(const TFormatHandlerStatistic& stat) {
+        ParseAndFilterLatency = stat.ParseAndFilterLatency != TDuration::Zero() ? stat.ParseAndFilterLatency : ParseAndFilterLatency;
+
+        ParserStats.Add(stat.ParserStats);
+        FilterStats.Add(stat.FilterStats);
+    }
+};
+
+struct TTopicSessionCommonStatistic {
     ui64 UnreadBytes = 0;   // Current value
     ui64 RestartSessionByOffsets = 0;
     ui64 ReadBytes = 0;     // Increment
     ui64 ReadEvents = 0;    // Increment
     ui64 LastReadedOffset = 0;
-    TDuration ParseAndFilterLatency;
-    void Add(const TopicSessionCommonStatistic& stat) {
+
+    std::unordered_map<TString, TFormatHandlerStatistic> FormatHandlers;
+
+    void Add(const TTopicSessionCommonStatistic& stat) {
         UnreadBytes = stat.UnreadBytes;
         RestartSessionByOffsets = stat.RestartSessionByOffsets;
         ReadBytes += stat.ReadBytes;
         ReadEvents += stat.ReadEvents;
         LastReadedOffset = stat.LastReadedOffset;
-        ParseAndFilterLatency = stat.ParseAndFilterLatency != TDuration::Zero() ? stat.ParseAndFilterLatency : ParseAndFilterLatency;
+
+        for (const auto& [formatName, foramtStats] : stat.FormatHandlers) {
+            FormatHandlers[formatName].Add(foramtStats);
+        }
     }
+
     void Clear() {
         ReadBytes = 0;
         ReadEvents = 0;
     }
 };
 
-struct TopicSessionParams {
+struct TTopicSessionParams {
     TString ReadGroup;
     TString Endpoint;
     TString Database;
     TString TopicPath;
     ui64 PartitionId = 0;
 };
 
-struct TopicSessionStatistic {
-    TopicSessionParams SessionKey; 
-    TVector<TopicSessionClientStatistic> Clients;
-    TopicSessionCommonStatistic Common;
+struct TTopicSessionStatistic {
+    TTopicSessionParams SessionKey; 
+    std::vector<TTopicSessionClientStatistic> Clients;
+    TTopicSessionCommonStatistic Common;
 };
 
 } // namespace NFq

ydb/core/fq/libs/row_dispatcher/events/ya.make

@@ -7,8 +7,11 @@ SRCS(
 PEERDIR(
     ydb/core/fq/libs/events
     ydb/core/fq/libs/row_dispatcher/protos
+
     ydb/library/actors/core
     ydb/library/yql/providers/pq/provider
+
+    yql/essentials/public/issue
 )
 
 END()

ydb/core/fq/libs/row_dispatcher/format_handler/common/common.cpp

@@ -0,0 +1,13 @@
+#include "common.h"
+
+#include <util/string/builder.h>
+
+namespace NFq::NRowDispatcher {
+
+//// TSchemaColumn
+
+TString TSchemaColumn::ToString() const {
+    return TStringBuilder() << "'" << Name << "' : " << TypeYson;
+}
+
+}  // namespace NFq::NRowDispatcher

ydb/core/fq/libs/row_dispatcher/format_handler/common/common.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <ydb/library/conclusion/generic/result.h>
+#include <ydb/library/conclusion/status.h>
+#include <ydb/library/yql/dq/actors/protos/dq_status_codes.pb.h>
+
+namespace NFq::NRowDispatcher {
+
+using EStatusId = NYql::NDqProto::StatusIds;
+using TStatusCode = EStatusId::StatusCode;
+using TStatus = NKikimr::TYQLConclusionSpecialStatus<TStatusCode, EStatusId::SUCCESS, EStatusId::INTERNAL_ERROR>;
+
+template <typename TValue>
+using TValueStatus = NKikimr::TConclusionImpl<TStatus, TValue>;
+
+struct TSchemaColumn {
+    TString Name;
+    TString TypeYson;
+
+    bool operator==(const TSchemaColumn& other) const = default;
+
+    TString ToString() const;
+};
+
+}  // namespace NFq::NRowDispatcher

ydb/core/fq/libs/row_dispatcher/format_handler/common/ya.make

@@ -0,0 +1,16 @@
+LIBRARY()
+
+SRCS(
+    common.cpp
+)
+
+PEERDIR(
+    ydb/library/conclusion
+    ydb/library/yql/dq/actors/protos
+
+    yql/essentials/public/issue
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()