YQ-4049 / YQ-4071 Sink fq to stable (#14018)

kardymonds · web-flow · commit 2dede2bdddc0 · 2025-02-03T20:01:36.000+03:00
diff --git a/ydb/core/fq/libs/checkpointing/checkpoint_coordinator.cpp b/ydb/core/fq/libs/checkpointing/checkpoint_coordinator.cpp
@@ -387,22 +387,29 @@ void TCheckpointCoordinator::Handle(const TEvCheckpointStorage::TEvCreateCheckpo
     const auto& issues = ev->Get()->Issues;
     CC_LOG_D("[" << checkpointId << "] Got TEvCreateCheckpointResponse");
 
-    if (issues) {
-        CC_LOG_E("[" << checkpointId << "] StorageError: can't create checkpoint: " << issues.ToOneLineString());
+    auto cancelCheckpoint = [&](const TString& str) {
+        CC_LOG_E("[" << checkpointId << "] " << str);
         PendingCheckpoints.erase(checkpointId);
         FailedZeroCheckpoint = InitingZeroCheckpoint;
         UpdateInProgressMetric();
         ++*Metrics.FailedToCreate;
         ++*Metrics.StorageError;
         CheckpointingSnapshotRotationIndex = CheckpointingSnapshotRotationPeriod; // Next checkpoint is snapshot.
+    };
+
+    if (issues) {
+        cancelCheckpoint("StorageError: can't create checkpoint: " + issues.ToOneLineString());
         return;
     }
 
     if (GraphDescId) {
         Y_ABORT_UNLESS(GraphDescId == ev->Get()->GraphDescId);
     } else {
         GraphDescId = ev->Get()->GraphDescId;
-        Y_ABORT_UNLESS(GraphDescId);
+        if (!GraphDescId) {
+            cancelCheckpoint("StorageError (internal error), empty GraphDescId");
+            return;
+        }
     }
 
     if (PendingInit) {
diff --git a/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.cpp b/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.cpp
@@ -457,6 +457,12 @@ class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public
         return statistics;
     }
 
+    void ForceRefresh() override {
+        if (Parser) {
+            Parser->Refresh(true);
+        }
+    }
+
 protected:
     NActors::TActorId GetSelfId() const override {
         return SelfId();
diff --git a/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.h b/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.h
@@ -55,6 +55,7 @@ class ITopicFormatHandler : public TNonCopyable {
     virtual bool HasClients() const = 0;
 
     virtual TFormatHandlerStatistic GetStatistics() = 0;
+    virtual void ForceRefresh() = 0;
 
 protected:
     virtual NActors::TActorId GetSelfId() const = 0;
diff --git a/ydb/core/fq/libs/row_dispatcher/topic_session.cpp b/ydb/core/fq/libs/row_dispatcher/topic_session.cpp
@@ -286,7 +286,8 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
     void SubscribeOnNextEvent();
     void SendToParsing(const TVector<NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage>& messages);
     void SendData(TClientsInfo& info);
-    void FatalError(TStatus status);
+    void FatalError(const TStatus& status);
+    void ThrowFatalError(const TStatus& status);
     void SendDataArrived(TClientsInfo& client);
     void StopReadSession();
     TString GetSessionId() const;
@@ -308,6 +309,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
     TMaybe<ui64> GetOffset(const NFq::NRowDispatcherProto::TEvStartSession& settings);
     void SendSessionError(TActorId readActorId, TStatus status);
     void RestartSessionIfOldestClient(const TClientsInfo& info);
+    void RefreshParsers();
 
 private:
 
@@ -330,7 +332,8 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
         IgnoreFunc(TEvRowDispatcher::TEvGetNextBatch);
         IgnoreFunc(NFq::TEvRowDispatcher::TEvStartSession);
         IgnoreFunc(NFq::TEvRowDispatcher::TEvStopSession);
-        IgnoreFunc(NFq::TEvPrivate::TEvSendStatistic);,
+        IgnoreFunc(NFq::TEvPrivate::TEvSendStatistic);
+        IgnoreFunc(NFq::TEvPrivate::TEvReconnectSession);,
         ExceptionFunc(std::exception, HandleException)
     )
 };
@@ -497,6 +500,7 @@ void TTopicSession::Handle(NFq::TEvPrivate::TEvReconnectSession::TPtr&) {
     LOG_ROW_DISPATCHER_DEBUG("Reconnect topic session, " << TopicPathPartition
         << ", StartingMessageTimestamp " << minTime
         << ", BufferSize " << BufferSize << ", WithoutConsumer " << Config.GetWithoutConsumer());
+    RefreshParsers();
     StopReadSession();
     CreateTopicSession();
     Schedule(ReconnectPeriod, new NFq::TEvPrivate::TEvReconnectSession());
@@ -565,7 +569,7 @@ void TTopicSession::TTopicEventProcessor::operator()(NYdb::NTopic::TSessionClose
     const TString message = TStringBuilder() << "Read session to topic \"" << Self.TopicPathPartition << "\" was closed";
     LOG_ROW_DISPATCHER_DEBUG(message << ": " << ev.DebugString());
 
-    Self.FatalError(TStatus::Fail(
+    Self.ThrowFatalError(TStatus::Fail(
         NYql::NDq::YdbStatusToDqStatus(static_cast<Ydb::StatusIds::StatusCode>(ev.GetStatus())),
         ev.GetIssues()
     ).AddParentIssue(message));
@@ -773,14 +777,15 @@ void TTopicSession::RestartSessionIfOldestClient(const TClientsInfo& info) {
     Metrics.RestartSessionByOffsets->Inc();
     ++RestartSessionByOffsets;
     info.RestartSessionByOffsetsByQuery->Inc();
+    RefreshParsers();
     StopReadSession();
 
     if (!ReadSession) {
         Schedule(TDuration::Seconds(Config.GetTimeoutBeforeStartSessionSec()), new NFq::TEvPrivate::TEvCreateSession());
     }
 }
 
-void TTopicSession::FatalError(TStatus status) {
+void TTopicSession::FatalError(const TStatus& status) {
     LOG_ROW_DISPATCHER_ERROR("FatalError: " << status.GetErrorMessage());
 
     for (auto& [readActorId, info] : Clients) {
@@ -789,7 +794,11 @@ void TTopicSession::FatalError(TStatus status) {
     }
     StopReadSession();
     Become(&TTopicSession::ErrorState);
-    ythrow yexception() << "FatalError: " << status.GetErrorMessage();    // To exit from current stack and call once PassAway() in HandleException().
+}
+
+void TTopicSession::ThrowFatalError(const TStatus& status) {
+    FatalError(status);
+    ythrow yexception() << "FatalError: " << status.GetErrorMessage();
 }
 
 void TTopicSession::SendSessionError(TActorId readActorId, TStatus status) {
@@ -902,6 +911,12 @@ TMaybe<ui64> TTopicSession::GetOffset(const NFq::NRowDispatcherProto::TEvStartSe
     return Nothing();
 }
 
+void TTopicSession::RefreshParsers() {
+    for (const auto& [_, formatHandler] : FormatHandlers) {
+        formatHandler->ForceRefresh();
+    }
+}
+
 }  // anonymous namespace
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/ydb/core/fq/libs/ydb/ut/ydb_ut.cpp b/ydb/core/fq/libs/ydb/ut/ydb_ut.cpp
@@ -466,6 +466,18 @@ Y_UNIT_TEST_SUITE(TFqYdbTest) {
         UNIT_ASSERT(issues.Size() == 1);
         UNIT_ASSERT(issues.ToString().Contains(text));
     }
+
+    Y_UNIT_TEST(ShouldStatusToIssuesProcessEmptyIssues)
+    {
+        auto promise = NThreading::NewPromise<NYdb::TStatus>();
+        auto future = promise.GetFuture();
+        promise.SetValue(TStatus(EStatus::BAD_REQUEST, NYql::TIssues{}));
+        NThreading::TFuture<NYql::TIssues> future2 = NFq::StatusToIssues(future);
+
+        NYql::TIssues issues = future2.GetValueSync();
+        UNIT_ASSERT_C(issues.Size() == 1, issues.ToString());
+        UNIT_ASSERT(issues.ToString().Contains("empty issues"));
+    }
 }
 
 } // namespace NFq
diff --git a/ydb/core/fq/libs/ydb/ydb.cpp b/ydb/core/fq/libs/ydb/ydb.cpp
@@ -203,6 +203,11 @@ NYql::TIssues StatusToIssues(const NYdb::TStatus& status) {
     TIssues issues;
     if (!status.IsSuccess()) {
         issues = status.GetIssues();
+        if (!issues) {
+            TStringStream str;
+            str << "Internal error: empty issues with failed status (" << status.GetStatus() << ")";
+            issues.AddIssue(str.Str());
+        }
     }
     return issues;
 }
diff --git a/ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.cpp b/ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.cpp
@@ -103,18 +103,21 @@ struct TEvPrivate {
         EvBegin = EventSpaceBegin(NActors::TEvents::ES_PRIVATE),
         EvPrintState = EvBegin + 20,
         EvProcessState = EvBegin + 21,
+        EvNotifyCA = EvBegin + 22,
         EvEnd
     };
     static_assert(EvEnd < EventSpaceEnd(NActors::TEvents::ES_PRIVATE), "expect EvEnd < EventSpaceEnd(NActors::TEvents::ES_PRIVATE)");
     struct TEvPrintState : public NActors::TEventLocal<TEvPrintState, EvPrintState> {};
     struct TEvProcessState : public NActors::TEventLocal<TEvProcessState, EvProcessState> {};
+    struct TEvNotifyCA : public NActors::TEventLocal<TEvNotifyCA, EvNotifyCA> {};
 };
 
 class TDqPqRdReadActor : public NActors::TActor<TDqPqRdReadActor>, public NYql::NDq::NInternal::TDqPqReadActorBase {
 
     const ui64 PrintStatePeriodSec = 300;
     const ui64 ProcessStatePeriodSec = 1;
     const ui64 PrintStateToLogSplitSize = 64000;
+    const ui64 NotifyCAPeriodSec = 10;
 
     struct TReadyBatch {
     public:
@@ -263,6 +266,7 @@ class TDqPqRdReadActor : public NActors::TActor<TDqPqRdReadActor>, public NYql::
     void Handle(const NFq::TEvRowDispatcher::TEvHeartbeat::TPtr&);
     void Handle(TEvPrivate::TEvPrintState::TPtr&);
     void Handle(TEvPrivate::TEvProcessState::TPtr&);
+    void Handle(TEvPrivate::TEvNotifyCA::TPtr&);
 
     STRICT_STFUNC(StateFunc, {
         hFunc(NFq::TEvRowDispatcher::TEvCoordinatorChanged, Handle);
@@ -283,6 +287,7 @@ class TDqPqRdReadActor : public NActors::TActor<TDqPqRdReadActor>, public NYql::
         hFunc(NFq::TEvRowDispatcher::TEvHeartbeat, Handle);
         hFunc(TEvPrivate::TEvPrintState, Handle);
         hFunc(TEvPrivate::TEvProcessState, Handle);
+        hFunc(TEvPrivate::TEvNotifyCA, Handle);
     })
 
     static constexpr char ActorName[] = "DQ_PQ_READ_ACTOR";
@@ -379,6 +384,7 @@ void TDqPqRdReadActor::Init() {
     Send(LocalRowDispatcherActorId, new NFq::TEvRowDispatcher::TEvCoordinatorChangesSubscribe());
 
     Schedule(TDuration::Seconds(PrintStatePeriodSec), new TEvPrivate::TEvPrintState());
+    Schedule(TDuration::Seconds(NotifyCAPeriodSec), new TEvPrivate::TEvNotifyCA());
     Inited = true;
 }
 
@@ -1004,6 +1010,11 @@ void TDqPqRdReadActor::UpdateSessions() {
     LastUsedPartitionDistribution = LastReceivedPartitionDistribution;
 }
 
+void TDqPqRdReadActor::Handle(TEvPrivate::TEvNotifyCA::TPtr&) {
+    Schedule(TDuration::Seconds(NotifyCAPeriodSec), new TEvPrivate::TEvNotifyCA());
+    NotifyCA();
+}
+
 std::pair<IDqComputeActorAsyncInput*, NActors::IActor*> CreateDqPqRdReadActor(
     const TTypeEnvironment& typeEnv,
     NPq::NProto::TDqPqTopicSource&& settings,

Original file line number	Diff line number	Diff line change
`@@ -203,6 +203,11 @@ NYql::TIssues StatusToIssues(const NYdb::TStatus& status) {`
`203`	`203`	`TIssues issues;`
`204`	`204`	`if (!status.IsSuccess()) {`
`205`	`205`	`issues = status.GetIssues();`
	`206`	`+ if (!issues) {`
	`207`	`+ TStringStream str;`
	`208`	`+ str << "Internal error: empty issues with failed status (" << status.GetStatus() << ")";`
	`209`	`+ issues.AddIssue(str.Str());`
	`210`	`+ }`
`206`	`211`	`}`
`207`	`212`	`return issues;`
`208`	`213`	`}`