Fix for TEvPut/GetResult (#6040)

mregrock · Egor Kulin · web-flow · commit 4fde7f62e6c7 · 2024-06-28T09:34:41.000+03:00
Co-authored-by: Egor Kulin &lt;mregrock@localhost.localdomain&gt;
diff --git a/ydb/core/base/blobstorage.h b/ydb/core/base/blobstorage.h
@@ -1001,7 +1001,7 @@ struct TEvBlobStorage {
         NKikimrProto::EReplyStatus Status;
         const TLogoBlobID Id;
         const TStorageStatusFlags StatusFlags;
-        const TGroupId GroupId;
+        const ui32 GroupId;
         const float ApproximateFreeSpaceShare; // 0.f has special meaning 'data could not be obtained'
         TString ErrorReason;
         bool WrittenBeyondBarrier = false; // was this blob written beyond the barrier?
@@ -1014,6 +1014,16 @@ struct TEvBlobStorage {
             : Status(status)
             , Id(id)
             , StatusFlags(statusFlags)
+            , GroupId(groupId.GetRawId())
+            , ApproximateFreeSpaceShare(approximateFreeSpaceShare)
+            , StorageId(storageId)
+        {}
+
+        TEvPutResult(NKikimrProto::EReplyStatus status, const TLogoBlobID &id, const TStorageStatusFlags statusFlags,
+                ui32 groupId, float approximateFreeSpaceShare, const TString& storageId = Default<TString>())
+            : Status(status)
+            , Id(id)
+            , StatusFlags(statusFlags)
             , GroupId(groupId)
             , ApproximateFreeSpaceShare(approximateFreeSpaceShare)
             , StorageId(storageId)
@@ -1231,7 +1241,7 @@ struct TEvBlobStorage {
         // todo: replace with queue-like thing
         ui32 ResponseSz;
         TArrayHolder<TResponse> Responses;
-        const TGroupId GroupId;
+        const ui32 GroupId;
         ui32 BlockedGeneration = 0; // valid only for requests with non-zero TabletId and true AcquireBlockedGeneration.
         TString DebugInfo;
         TString ErrorReason;
@@ -1242,6 +1252,13 @@ struct TEvBlobStorage {
         TInstant Sent;
 
         TEvGetResult(NKikimrProto::EReplyStatus status, ui32 sz, TGroupId groupId)
+            : Status(status)
+            , ResponseSz(sz)
+            , Responses(sz == 0 ? nullptr : new TResponse[sz])
+            , GroupId(groupId.GetRawId())
+        {}
+
+        TEvGetResult(NKikimrProto::EReplyStatus status, ui32 sz, ui32 groupId)
             : Status(status)
             , ResponseSz(sz)
             , Responses(sz == 0 ? nullptr : new TResponse[sz])
diff --git a/ydb/core/blobstorage/dsproxy/dsproxy_get.cpp b/ydb/core/blobstorage/dsproxy/dsproxy_get.cpp
@@ -368,7 +368,7 @@ class TBlobStorageGroupGetRequest : public TBlobStorageGroupRequestActor<TBlobSt
         LWTRACK(DSProxyGetReply, Orbit);
         evResult->Orbit = std::move(Orbit);
         LWPROBE(DSProxyRequestDuration, TEvBlobStorage::EvGet, requestSize, duration.SecondsFloat() * 1000.0, tabletId,
-                evResult->GroupId.GetRawId(), channel, NKikimrBlobStorage::EGetHandleClass_Name(GetImpl.GetHandleClass()),
+                evResult->GroupId, channel, NKikimrBlobStorage::EGetHandleClass_Name(GetImpl.GetHandleClass()),
                 success);
         A_LOG_LOG_S(true, success ? NLog::PRI_INFO : NLog::PRI_NOTICE, "BPG68", "Result# " << evResult->Print(false));
         return SendResponseAndDie(std::unique_ptr<TEvBlobStorage::TEvGetResult>(evResult.Release()));
diff --git a/ydb/core/blobstorage/testing/group_overseer/group_overseer.cpp b/ydb/core/blobstorage/testing/group_overseer/group_overseer.cpp
@@ -93,7 +93,11 @@ namespace NKikimr::NTesting {
             QueryToGroup.erase(it);
 
             auto& msg = *ev.Get<T>();
-            if constexpr (T::EventType != TEvBlobStorage::EvBlockResult &&
+
+            if constexpr (T::EventType == TEvBlobStorage::EvGetResult || T::EventType == TEvBlobStorage::EvPutResult){
+                Y_ABORT_UNLESS(groupId == msg.GroupId);
+            }
+            else if constexpr (T::EventType != TEvBlobStorage::EvBlockResult &&
                     T::EventType != TEvBlobStorage::EvInplacePatchResult &&
                     T::EventType != TEvBlobStorage::EvCollectGarbageResult &&
                     T::EventType != TEvBlobStorage::EvDiscoverResult) {
diff --git a/ydb/core/keyvalue/keyvalue_storage_read_request.cpp b/ydb/core/keyvalue/keyvalue_storage_read_request.cpp
@@ -226,11 +226,11 @@ class TKeyValueStorageReadRequest : public TActorBootstrapped<TKeyValueStorageRe
 
         TGetBatch &batch = Batches[ev->Cookie];
 
-        if (result->GroupId.GetRawId() != batch.GroupId) {
+        if (result->GroupId != batch.GroupId) {
             STLOG_WITH_ERROR_DESCRIPTION(ErrorDescription, NLog::PRI_ERROR, NKikimrServices::KEYVALUE, KV318,
                     "Received EvGetResult from an unexpected storage group.",
                     (KeyValue, TabletInfo->TabletID),
-                    (GroupId, result->GroupId.GetRawId()),
+                    (GroupId, result->GroupId),
                     (ExpecetedGroupId, batch.GroupId),
                     (Status, result->Status),
                     (Deadline, IntermediateResult->Deadline.MilliSeconds()),
diff --git a/ydb/core/keyvalue/keyvalue_storage_request.cpp b/ydb/core/keyvalue/keyvalue_storage_request.cpp
@@ -109,7 +109,7 @@ class TKeyValueStorageRequest : public TActorBootstrapped<TKeyValueStorageReques
         IntermediateResults->Stat.PutLatencies.push_back(duration.MilliSeconds());
 
         auto groupId = ev->Get()->GroupId;
-        CheckYellow(ev->Get()->StatusFlags, groupId.GetRawId());
+        CheckYellow(ev->Get()->StatusFlags, groupId);
 
         NKikimrProto::EReplyStatus status = ev->Get()->Status;
         if (status != NKikimrProto::OK) {
@@ -157,8 +157,8 @@ class TKeyValueStorageRequest : public TActorBootstrapped<TKeyValueStorageReques
         wr->StatusFlags.Merge(ev->Get()->StatusFlags.Raw);
         wr->Latency = duration;
         ++WriteRequestsReplied;
-        IntermediateResults->Stat.GroupWrittenBytes[std::make_pair(ev->Get()->Id.Channel(), groupId.GetRawId())] += ev->Get()->Id.BlobSize();
-        IntermediateResults->Stat.GroupWrittenIops[std::make_pair(ev->Get()->Id.Channel(), groupId.GetRawId())] += 1; // FIXME: count distinct blobs?
+        IntermediateResults->Stat.GroupWrittenBytes[std::make_pair(ev->Get()->Id.Channel(), groupId)] += ev->Get()->Id.BlobSize();
+        IntermediateResults->Stat.GroupWrittenIops[std::make_pair(ev->Get()->Id.Channel(), groupId)] += 1; // FIXME: count distinct blobs?
         UpdateRequest(ctx);
     }
 
@@ -337,8 +337,8 @@ class TKeyValueStorageRequest : public TActorBootstrapped<TKeyValueStorageReques
             if (response.Status == NKikimrProto::OK) {
                 Y_ABORT_UNLESS(response.Buffer.size() == readItem.BlobSize);
                 Y_ABORT_UNLESS(readItem.ValueOffset + readItem.BlobSize <= read.ValueSize);
-                IntermediateResults->Stat.GroupReadBytes[std::make_pair(response.Id.Channel(), groupId.GetRawId())] += response.Buffer.size();
-                IntermediateResults->Stat.GroupReadIops[std::make_pair(response.Id.Channel(), groupId.GetRawId())] += 1; // FIXME: count distinct blobs?
+                IntermediateResults->Stat.GroupReadBytes[std::make_pair(response.Id.Channel(), groupId)] += response.Buffer.size();
+                IntermediateResults->Stat.GroupReadIops[std::make_pair(response.Id.Channel(), groupId)] += 1; // FIXME: count distinct blobs?
                 read.Value.Write(readItem.ValueOffset, std::move(response.Buffer));
             } else {
                 Y_VERIFY_DEBUG_S(response.Status != NKikimrProto::NODATA, "NODATA received for TEvGet"
diff --git a/ydb/core/tablet/tablet_req_rebuildhistory.cpp b/ydb/core/tablet/tablet_req_rebuildhistory.cpp
@@ -589,8 +589,8 @@ class TTabletReqRebuildHistoryGraph : public TActorBootstrapped<TTabletReqRebuil
             switch (response.Status) {
             case NKikimrProto::OK:
                 Y_ABORT_UNLESS(1 == RefsToCheck.erase(response.Id));
-                GroupReadBytes[std::make_pair(response.Id.Channel(), msg->GroupId.GetRawId())] += response.Buffer.size();
-                GroupReadOps[std::make_pair(response.Id.Channel(), msg->GroupId.GetRawId())] += 1;
+                GroupReadBytes[std::make_pair(response.Id.Channel(), msg->GroupId)] += response.Buffer.size();
+                GroupReadOps[std::make_pair(response.Id.Channel(), msg->GroupId)] += 1;
                 break;
             case NKikimrProto::NODATA:
                 BLOG_W("TTabletReqRebuildHistoryGraph::CheckReferences - NODATA for blob " << response.Id, "TRRH07");
diff --git a/ydb/core/tablet/tablet_req_writelog.cpp b/ydb/core/tablet/tablet_req_writelog.cpp
@@ -49,8 +49,8 @@ class TTabletReqWriteLog : public TActorBootstrapped<TTabletReqWriteLog> {
         case NKikimrProto::OK:
             LOG_DEBUG_S(ctx, NKikimrServices::TABLET_MAIN, "Put Result: " << msg->Print(false));
 
-            GroupWrittenBytes[std::make_pair(msg->Id.Channel(), msg->GroupId.GetRawId())] += msg->Id.BlobSize();
-            GroupWrittenOps[std::make_pair(msg->Id.Channel(), msg->GroupId.GetRawId())] += 1;
+            GroupWrittenBytes[std::make_pair(msg->Id.Channel(), msg->GroupId)] += msg->Id.BlobSize();
+            GroupWrittenOps[std::make_pair(msg->Id.Channel(), msg->GroupId)] += 1;
 
             ResponseCookies ^= ev->Cookie;
 
diff --git a/ydb/core/tablet_flat/flat_ops_compact.h b/ydb/core/tablet_flat/flat_ops_compact.h
@@ -454,7 +454,7 @@ namespace NTabletFlatExecutor {
             }
 
             if (ok) {
-                Send(Owner, new NBlockIO::TEvStat(NBlockIO::EDir::Write, NBlockIO::EPriority::Bulk, msg.GroupId.GetRawId(), msg.Id));
+                Send(Owner, new NBlockIO::TEvStat(NBlockIO::EDir::Write, NBlockIO::EPriority::Bulk, msg.GroupId, msg.Id));
 
                 while (!WriteQueue.empty() && Writing < MaxFlight) {
                     SendToBs(std::move(WriteQueue.front()));
diff --git a/ydb/core/tx/columnshard/engines/writer/write_controller.cpp b/ydb/core/tx/columnshard/engines/writer/write_controller.cpp
@@ -4,7 +4,7 @@
 namespace NKikimr::NColumnShard {
 
 void IWriteController::OnBlobWriteResult(const TEvBlobStorage::TEvPutResult& result) {
-    NOlap::TUnifiedBlobId blobId(result.GroupId.GetRawId(), result.Id);
+    NOlap::TUnifiedBlobId blobId(result.GroupId, result.Id);
     auto it = WaitingActions.find(result.StorageId ? result.StorageId : NOlap::IStoragesManager::DefaultStorageId);
     AFL_VERIFY(it != WaitingActions.end());
     it->second->OnBlobWriteResult(blobId, result.Status);

Original file line number	Diff line number	Diff line change
`@@ -454,7 +454,7 @@ namespace NTabletFlatExecutor {`
`454`	`454`	`}`
`455`	`455`
`456`	`456`	`if (ok) {`
`457`		`- Send(Owner, new NBlockIO::TEvStat(NBlockIO::EDir::Write, NBlockIO::EPriority::Bulk, msg.GroupId.GetRawId(), msg.Id));`
	`457`	`+ Send(Owner, new NBlockIO::TEvStat(NBlockIO::EDir::Write, NBlockIO::EPriority::Bulk, msg.GroupId, msg.Id));`
`458`	`458`
`459`	`459`	`while (!WriteQueue.empty() && Writing < MaxFlight) {`
`460`	`460`	`SendToBs(std::move(WriteQueue.front()));`