Don't create batch in TBlobIterator if it's not necessary.

![flamegraph](https://arcanum.s3.mds.yandex.net/files/azevaykin/LtGVsw2KQ7J7uzglcw2a5)
Функция CheckBlob в режиме проверки  выполняет лишние создания TBatch, который потом не используется.
Введен флаг  bool createBatch

2 files changed, 29 insertions, 18 deletions

diff --git a/ydb/core/persqueue/blob.cpp b/ydb/core/persqueue/blob.cpp
index 21d687a5e42..b35d4554490 100644
--- a/ydb/core/persqueue/blob.cpp
+++ b/ydb/core/persqueue/blob.cpp
@@ -8,19 +8,19 @@
 namespace NKikimr {
 namespace NPQ {
 
-
-TBlobIterator::TBlobIterator(const TKey& key, const TString& blob)
-: Key(key)
-, Data(blob.c_str())
-, End(Data + blob.size())
-, Batch()
-, Offset(key.GetOffset())
-, Count(0)
-, InternalPartsCount(0)
-{
-    Y_VERIFY(Data != End);
-    ParseBatch(true);
-}
+    TBlobIterator::TBlobIterator(const TKey& key, const TString& blob, bool createBatch)
+        : CreateBatch(createBatch)
+        , Batch()
+        , Key(key)
+        , Data(blob.c_str())
+        , End(Data + blob.size())
+        , Offset(key.GetOffset())
+        , Count(0)
+        , InternalPartsCount(0)
+    {
+        Y_VERIFY(Data != End);
+        ParseBatch(true);
+    }
 
 void TBlobIterator::ParseBatch(bool isFirst) {
     Y_VERIFY(Data < End);
@@ -34,7 +34,10 @@ void TBlobIterator::ParseBatch(bool isFirst) {
     Y_VERIFY(Count <= Key.GetCount());
     Y_VERIFY(InternalPartsCount <= Key.GetInternalPartsCount());
 
-    Batch = TBatch(header, Data + sizeof(ui16) + header.ByteSize());
+    if(CreateBatch)
+        Batch = TBatch(header, Data + sizeof(ui16) + header.ByteSize());
+    else
+        Header = std::move(header);
 }
 
 bool TBlobIterator::IsValid()
@@ -45,7 +48,7 @@ bool TBlobIterator::IsValid()
 bool TBlobIterator::Next()
 {
     Y_VERIFY(IsValid());
-    auto& header = Batch.Header;
+    NKikimrPQ::TBatchHeader& header = CreateBatch ? Batch.Header : Header;
     Data += header.GetPayloadSize() + sizeof(ui16) + header.ByteSize();
     if (Data == End) { //this was last batch
         Y_VERIFY(Count == Key.GetCount());
@@ -58,13 +61,14 @@ bool TBlobIterator::Next()
 
 const TBatch& TBlobIterator::GetBatch()
 {
+    Y_VERIFY(CreateBatch);
     Y_VERIFY(IsValid());
     return Batch;
 }
 
 void CheckBlob(const TKey& key, const TString& blob)
 {
-    for (TBlobIterator it(key, blob); it.IsValid(); it.Next());
+    for (TBlobIterator it(key, blob, false); it.IsValid(); it.Next());
 }
 
 
diff --git a/ydb/core/persqueue/blob.h b/ydb/core/persqueue/blob.h
index 633ec2499a3..244349edfed 100644
--- a/ydb/core/persqueue/blob.h
+++ b/ydb/core/persqueue/blob.h
@@ -197,7 +197,8 @@ struct TBatch {
 
 class TBlobIterator {
 public:
-    TBlobIterator(const TKey& key, const TString& blob);
+    TBlobIterator(const TKey& key, const TString& blob, bool createBatch = true);
+
     //return true is there is batch
     bool IsValid();
     //get next batch and return false if there is no next batch
@@ -207,10 +208,16 @@ public:
 private:
     void ParseBatch(bool isFirst);
 
+    // if true, Batch is filled, otherwise only Header.
+    bool CreateBatch;
+
+    TBatch Batch;
+    NKikimrPQ::TBatchHeader Header;
+
     const TKey& Key;
     const char *Data;
     const char *End;
-    TBatch Batch;
+
     ui64 Offset;
     ui32 Count;
     ui16 InternalPartsCount;