YQ-3981 RD fixed missing historical data (#12757)

Author: GrigoriyPA

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

@@ -348,6 +348,10 @@ class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public
     void ParseMessages(const TVector<NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage>& messages) override {
         LOG_ROW_DISPATCHER_TRACE("Send " << messages.size() << " messages to parser");
 
+        if (messages) {
+            CurrentOffset = messages.back().GetOffset();
+        }
+
         if (Parser) {
             Parser->ParseMessages(messages);
             ScheduleRefresh();
@@ -367,6 +371,13 @@ class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public
     TStatus AddClient(IClientDataConsumer::TPtr client) override {
         LOG_ROW_DISPATCHER_DEBUG("Add client with id " << client->GetClientId());
 
+        if (const auto clientOffset = client->GetNextMessageOffset()) {
+            if (Parser && CurrentOffset && *CurrentOffset > *clientOffset) {
+                LOG_ROW_DISPATCHER_DEBUG("Parser was flushed due to new historical offset " << *clientOffset << "(previous parser offset: " << *CurrentOffset << ")");
+                Parser->Refresh(true);
+            }
+        }
+
         auto clientHandler = MakeIntrusive<TClientHandler>(*this, client);
         if (!Clients.emplace(client->GetClientId(), clientHandler).second) {
             return TStatus::Fail(EStatusId::INTERNAL_ERROR, TStringBuilder() << "Failed to create new client, client with id " << client->GetClientId() << " already exists");
@@ -553,6 +564,7 @@ class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public
     ITopicParser::TPtr Parser;
     TParserHandler::TPtr ParserHandler;
     ITopicFilters::TPtr Filters;
+    std::optional<ui64> CurrentOffset;
 
     // Parsed data
     const TVector<ui64>* Offsets;

ydb/core/fq/libs/row_dispatcher/format_handler/parsers/json_parser.cpp

@@ -42,9 +42,15 @@ struct TJsonParserBuffer {
 
     void AddMessage(const NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage& message) {
         Y_ENSURE(!Finished, "Cannot add messages into finished buffer");
+
+        const auto offset = message.GetOffset();
+        if (Y_UNLIKELY(Offsets && Offsets.back() > offset)) {
+            LOG_ROW_DISPATCHER_WARN("Got message with offset " << offset << " which is less than previous offset " << Offsets.back());
+        }
+
         NumberValues++;
         Values << message.GetData();
-        Offsets.emplace_back(message.GetOffset());
+        Offsets.emplace_back(offset);
     }
 
     std::pair<const char*, size_t> Finish() {
@@ -65,6 +71,7 @@ struct TJsonParserBuffer {
 
 private:
     TStringBuilder Values = {};
+    const TString LogPrefix = "TJsonParser: Buffer: ";
 };
 
 class TColumnParser {